在Spark/PySpark中使用文件名加入数据

Question

我正在从PySpark中的多个S3文件中读取数据。 S3键包含创建文件的日历日期，我想在数据和该日期之间进行连接。有没有办法在文件和文件名中的数据行之间进行连接？

Answer 1

这里是我落得这样做：

我重写了LineRecordReader Hadoop类，以便它包含每行中的文件名，然后重写TextInputFormat以使用我的新LineRecordReader。

然后我使用newAPIHadoopFile函数加载文件。

链接：
LineRecordReader：http://tinyurl.com/linerecordreader
的TextInputFormat：http://tinyurl.com/textinputformat
newAPIHadoopFile： “欢迎来到SO” http://tinyurl.com/newapihadoopfile

Answer 2

您可以添加一列包含文件名，我用这个后来将它们合并后，以确定各行的源数据帧：

from pyspark.sql.functions import lit 

filename = 'myawesomefile.csv' 

df_new = df.withColumn('file_name', lit(filename))