是否可以在 pySpark 中修改输出数据文件名？答案

【问题标题】：Is it possible to modify output data file names in pySpark?是否可以在 pySpark 中修改输出数据文件名？
【发布时间】：2020-11-18 00:26:52
【问题描述】：

简化案例。

鉴于我在目录data_directory 中有 5 个输入文件：

data_2020-01-01.txt,
data_2020-01-02.txt,
data_2020-01-03.txt,
data_2020-01-04.txt,
data_2020-01-05.txt

我将它们全部读入 pySpark RDD 并对它们执行一些不做任何改组的操作。

spark = SparkSession.builder.appName("Clean Data").getOrCreate()
sparkContext = spark.sparkContext

input_rdd = sparkContext.textFile("data_directory/*")
result = input_rdd.mapPartitions(lambda x: remove_corrupted_rows(x))

现在我要保存数据：

result.saveAsTextFile(
    "results",
    compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec",
)

我得到 5 个文件，每个文件都包含名称“part”。所以我丢失了关于输出文件来自哪个输入文件的信息：

._SUCCESS.crc
.part-00000.gz.crc
.part-00001.gz.crc
.part-00002.gz.crc
.part-00003.gz.crc
.part-00004.gz.crc
_SUCCESS
part-00000.gz
part-00001.gz
part-00002.gz
part-00003.gz
part-00004.gz

在这种情况下是否有保留输入文件名或引入我自己的命名模式？

预期的结果：

._SUCCESS.crc
.data_2020-01-01.gz.crc
.data_2020-01-02.gz.crc
.data_2020-01-03.gz.crc
.data_2020-01-04.gz.crc
.data_2020-01-05.crc
_SUCCESS
data_2020-01-01.gz
data_2020-01-02.gz
data_2020-01-03.gz
data_2020-01-04.gz
data_2020-01-05.gz

【问题讨论】：

您的零件文件不保证包含这些日期（也不保证输出文件的数量相同）。如果要保留该信息，可以使用命名目录分区

标签： python apache-spark pyspark

【解决方案1】：

您可以使用pyspark.sql.functions.input_file_name()（此处为文档https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=functions#pyspark.sql.functions.input_file_name），然后按创建的列对数据框进行分区。

这样，5 个输入文件应该为您提供一个具有 5 个不同值的分类列，并且对其进行分区应该将您的输出分成 5 个部分。

或者，如果您希望拥有完整的命名模式，则在功能上拆分 input_file_name() 列上的数据框（这里分为 5 个数据框），重新分区（例如，使用 coalesce(1) 到 1），然后使用自定义逻辑保存（例如 dict 映射或通过从列中提取文件名并解析为 DataFrameWriter.csv() 作为名称）。

注意：当更改为 1 个分区时，请确保数据完全适合您的内存！

【讨论】：