将每个文件激发到数据集行

Question

我在一个目录中有许多文件，每个文件包含跨多行的文本。目前，我使用以下代码将所有这些文件读取到spark数据集(

   val ddf = spark.read.text("file:///input/*")

但是，这会创建一个数据集，其中每一行都是一行，而不是一个文件。我希望数据集中每行都有一个文件（作为字符串）。

如何在不迭代每个文件并将其作为RDD单独读取的情况下实现这一点？

Answer 1

@mrsrinivas的答案的另一种选择是按输入文件名分组。鉴于结构：

evan@vbox>~/junk/so> find .        
.
./d2
./d2/t.txt
./d1
./d1/t.txt
 evan@vbox>~/junk/so> cat  */*.txt
d1_1
d1_2
d2_1
d2_2

我们可以根据输入文件收集列表，如下所示：

scala prettyprint-override">scala> val ddf = spark.read.textFile("file:///home/evan/junk/so/*").
     | select($"value", input_file_name as "fName")
ddf: org.apache.spark.sql.DataFrame = [value: string, fName: string]

scala> ddf.show(false)
+-----+----------------------------------+
|value|fName                             |
+-----+----------------------------------+
|d2_1 |file:///home/evan/junk/so/d2/t.txt|
|d2_2 |file:///home/evan/junk/so/d2/t.txt|
|d1_1 |file:///home/evan/junk/so/d1/t.txt|
|d1_2 |file:///home/evan/junk/so/d1/t.txt|
+-----+----------------------------------+

scala> ddf.groupBy("fName").agg(collect_list($"value") as "value").
     | drop("fName").show
+------------+
|       value|
+------------+
|[d1_1, d1_2]|
|[d2_1, d2_2]|
+------------+

Answer 2

在SparkContext上使用wholeTextFiles（）

val rdd: RDD[(String, String)] = spark.sparkContext
                                      .wholeTextFiles("file/path/to/read/as/rdd")

SparkContext。wholeTextFiles允许您读取包含多个小文本文件的目录，并以（文件名、内容）对的形式返回每个小文本文件。这与textFile不同，textFile在每个文件中每行返回一条记录。