scala - 如何将源文件名添加到 Spark 中的每一行？

Question

我是 Spark 的新手，正在尝试使用它来自的文件名向每个输入行插入一列。

我见过其他人问过类似的问题，但他们所有的答案都使用 wholeTextFile ，但我正在尝试为较大的 CSV 文件(使用 Spark-CSV 库读取)、JSON 文件和 Parquet 文件(不仅仅是小文本文件)执行此操作。

我可以使用 spark-shell获取文件名列表:

val df = sqlContext.read.parquet("/blah/dir")
val names = df.select(inputFileName())
names.show

但这是一个数据框。
我不确定如何将它作为一列添加到每一行(如果该结果的排序也与初始数据相同，尽管我认为它总是如此)以及如何将其作为所有输入类型的通用解决方案。

Answer 1

我刚刚发现将文件名添加为 DataFrame 中的列之一的另一种解决方案

val df = sqlContext.read.parquet("/blah/dir")

val dfWithCol = df.withColumn("filename",input_file_name())

引用:
spark load data and add filename as dataframe column