arrays - 从Spark Scala的DataFrame列中的数组添加文件名

Question

val hadoopConf = new Configuration()

val fs = FileSystem.get(hadoopConf)

val status = fs.listStatus(new Path("/home/Test/")).map(_.getPath().toString)

val df =  spark.read.format("json").load(status : _*)

如何在df的新列中添加文件名？

我试过了:

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

但是它在所有列中添加了相同的文件名吗？
谁能建议一种更好的方法？

Answer 1

这是预期的行为，因为您的json文件中包含more than one record。

Spark为每条记录添加filenames，您要检查所有唯一的文件名，然后在文件名列上执行distinct

//to get unique filenames
df.select("filename").distinct().show()

Example:

#source data
hadoop fs -cat /user/shu/json/*.json
{"id":1,"name":"a"}
{"id":1,"name":"a"}

val hadoopConf = new Configuration()

val fs = FileSystem.get(hadoopConf)

val status = fs.listStatus(new Path("/user/shu/json")).map(_.getPath().toString)

val df =  spark.read.format("json").load(status : _*)

df.withColumn("filename",input_file_name()).show(false)

//unique filenames for each record
+---+----+----------------------------------------------------------------------------+
|id |name|input                                                                       |
+---+----+----------------------------------------------------------------------------+
|1  |a   |hdfs://nn:8020/user/shu/json/i.json                                         |
|1  |a   |hdfs://nn:8020/user/shu/json/i1.json                                        |
+---+----+----------------------------------------------------------------------------+

在上面的示例中，您可以看到每个记录 的unique filenames (as i have 1 record in each json file) 。