java - 如何使用文件中的列和字段创建 DataFrame？

Question

我必须创建一个带有标题和字段的 DataFrame。 header 和字段位于文件中。该文件指定如下。架构位于 field5 中，col1,col2...是我的架构，值位于 field6 之后。

field1 value1;
field2 value2;
field3 value3;
field4 value4;
field5 17 col1 col2 col3 col4 col5 col6 col7 col8;
field6
val1 val 2 val3 val4 val5 val6 val7 val8
val9 val10 val11 val12 val13 val14 val15 val16
val17 val18 val19 val20 val21 val22 val23 val24;
EndOfFile; 

上面是文件，我想提取值 col1,col2.......col8 并从中创建一个结构，并使用 field6 之后的值创建一个数据帧。

我应该使用普通的 Java 代码提取 field5 吗？ Spark Java 可以做到吗？

Answer 1

我会执行以下操作(但我使用的是 Scala，因此将其转换为 Java 是您的家庭练习):

1. 使用 spark.read.text 将文件加载为常规(几乎非结构化)文本文件
2. 过滤掉不相关的行
3. 使用请求的架构和行创建另一个 DataFrame

让我们看看 Scala 代码:

val input = spark.read.text("input.txt")
scala> input.show(false)
+--------------------------------------------------+
|value                                             |
+--------------------------------------------------+
|field1 value1;                                    |
|field2 value2;                                    |
|field3 value3;                                    |
|field4 value4;                                    |
|field5 17 col1 col2 col3 col4 col5 col6 col7 col8;|
|field6                                            |
|val1 val 2 val3 val4 val5 val6 val7 val8          |
|val9 val10 val11 val12 val13 val14 val15 val16    |
|val17 val18 val19 val20 val21 val22 val23 val24;  |
|EndOfFile;                                        |
+--------------------------------------------------+

// trying to impress future readers ;-)
val unnecessaryLines = (2 to 4).
  map(n => 'value startsWith s"field$n").
  foldLeft('value startsWith "field1") { case (f, orfield) => f or orfield }.
  or('value startsWith "field6").
  or('value startsWith "EndOfFile")
scala> unnecessaryLines.explain(true)
((((StartsWith('value, field1) || StartsWith('value, field2)) || StartsWith('value, field3)) || StartsWith('value, field4)) || StartsWith('value, EndOfFile))

// Filter out the irrelevant lines
val onlyRelevantLines = input.filter(!unnecessaryLines)
scala> onlyRelevantLines.show(false)
+--------------------------------------------------+
|value                                             |
+--------------------------------------------------+
|field5 17 col1 col2 col3 col4 col5 col6 col7 col8;|
|val1 val 2 val3 val4 val5 val6 val7 val8          |
|val9 val10 val11 val12 val13 val14 val15 val16    |
|val17 val18 val19 val20 val21 val22 val23 val24;  |
+--------------------------------------------------+

这样我们就从文件中获得了唯一相关的行。 是时候找点乐子了!

// Remove field5 from the first line only and `;` at the end
val field5 = onlyRelevantLines.head.getString(0) // we're leaving Spark space and enter Scala
// the following is pure Scala code (no Spark whatsoever)
val header = field5.substring("field5 17 ".size).dropRight(1).split("\\s+").toSeq

val rows = onlyRelevantLines.filter(!('value startsWith "field5"))
scala> :type rows
org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
scala> rows.show(false)
+------------------------------------------------+
|value                                           |
+------------------------------------------------+
|val1 val 2 val3 val4 val5 val6 val7 val8        |
|val9 val10 val11 val12 val13 val14 val15 val16  |
|val17 val18 val19 val20 val21 val22 val23 val24;|
+------------------------------------------------+

这样，您就有了一个应该分割(每个空格)的行的数据集。在未发布的 Spark 2.2.0 中，将有一个方法 csv这将加载数据集，并且给定的分隔符将为我们提供我们想要的内容:

def csv(csvDataset: Dataset[String]): DataFrame

目前尚不可用，因此我们必须做类似的事情。

让我们尽可能坚持使用 Spark SQL 的数据集 API。

val words = rows.select(split($"value", "\\s+") as "words")
scala> words.show(false)
+---------------------------------------------------------+
|words                                                    |
+---------------------------------------------------------+
|[val1, val, 2, val3, val4, val5, val6, val7, val8]       |
|[val9, val10, val11, val12, val13, val14, val15, val16]  |
|[val17, val18, val19, val20, val21, val22, val23, val24;]|
+---------------------------------------------------------+

// The following is just a series of withColumn's for every column in header

val finalDF = header.zipWithIndex.foldLeft(words) { case (df, (hdr, idx)) =>
  df.withColumn(hdr, $"words".getItem(idx)) }.
  drop("words")
scala> finalDF.show
+-----+-----+-----+-----+-----+-----+-----+------+
| col1| col2| col3| col4| col5| col6| col7|  col8|
+-----+-----+-----+-----+-----+-----+-----+------+
| val1|  val|    2| val3| val4| val5| val6|  val7|
| val9|val10|val11|val12|val13|val14|val15| val16|
|val17|val18|val19|val20|val21|val22|val23|val24;|
+-----+-----+-----+-----+-----+-----+-----+------+

完成!