如何在PySpark ETL中配置输出文件的大小、数量及单文件行数？

嗨，我来帮你搞定这个问题！你遇到的多文件输出其实是PySpark分布式计算的正常表现——RDD的每个分区最终会对应一个输出文件。要控制输出文件的数量、大小或者单文件行数，咱们可以从这几个方向入手：

一、直接控制输出文件数量（调整分区数） #

既然每个分区对应一个输出文件，那调整RDD的分区数就能直接控制最终生成的文件数量。这里有两个常用方法：

• coalesce(numPartitions)：适合减少分区数，它会尽量合并现有分区，避免大量数据shuffle（数据跨节点传输），性能更好。比如你想输出3个文件：

  merged_rdd.coalesce(3).saveAsTextFile('/output.json')
  

• repartition(numPartitions)：可以任意调整分区数（增加或减少），但会触发全量数据shuffle，适合需要重新均匀分配数据的场景。比如想输出5个文件：

  merged_rdd.repartition(5).saveAsTextFile('/output.json')
  

二、精准控制单文件行数 #

如果想让每个输出文件的行数尽量一致，可以通过给数据加索引，再按索引分区的方式实现：

1. 先给RDD的每条数据加上唯一索引
2. 根据你想要的单文件行数，计算需要的分区数
3. 按索引范围分配分区，最后去掉索引保存

示例代码：

# 给每条数据添加索引
indexed_rdd = merged_rdd.zipWithIndex()

# 设定每个文件的目标行数
rows_per_file = 2000
# 计算需要的分区数（向上取整）
total_rows = merged_rdd.count()
num_partitions = (total_rows + rows_per_file - 1) // rows_per_file

# 根据索引分配到对应分区
partitioned_rdd = indexed_rdd.partitionBy(
    num_partitions,
    lambda idx: idx // rows_per_file  # 按索引范围分区
)

# 去掉索引，保存结果
partitioned_rdd.map(lambda x: x[0]).saveAsTextFile('/output.json')

三、间接控制文件大小（通过行数限制） #

PySpark没有直接设置文件大小的参数，但可以通过限制单文件行数来间接控制大小。如果把RDD转换成DataFrame，还能利用Spark的内置配置参数：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.getOrCreate()

# 将RDD转换为DataFrame
result_df = spark.createDataFrame(merged_rdd)

# 设置每个输出文件的最大行数
spark.conf.set("spark.sql.files.maxRecordsPerFile", 2000)

# 保存DataFrame为文本文件
result_df.write.text('/output.json')

这个参数会确保每个输出文件的行数不超过你设定的值，间接帮你控制文件大小。

小提醒：尽量用PySpark原生API处理数据 #

看你的代码里先用了Pandas读CSV，其实如果数据量较大，建议换成PySpark的spark.read.csv，这样能更好地利用分布式计算能力，后续的分区和输出控制也更灵活：

df_one = spark.read.csv('one.csv', header=True, inferSchema=True)
df_two = spark.read.csv('two.csv', header=True, inferSchema=True)

# 用PySpark的join替代Pandas的merge
df_combined = df_one.join(df_two, df_one.truncated_id == df_two.id, how='inner')

备注：内容来源于stack exchange，提问作者tisha