如何配置Spark将小文件写入S3时生成多分区文件？

我来帮你搞定这个问题！你当前代码里有几个关键问题导致无法生成多个part文件，我一步步给你拆解并修正：

核心问题分析 #

1. coalesce用错了，且没生效：coalesce()是用来减少分区数的（它尽量避免数据洗牌，只能在特定场景下增加分区，非常不稳定），而且Spark的转换操作是懒加载的，你没有把coalesce(5)的结果赋值给新变量，这行代码等于白写！
2. 配置参数不对症：spark.files.maxPartitionBytes是控制读取文件时的分区大小，对写入时的part文件数量影响不大，关键还是要调整DataFrame的分区数。

具体修改方案 #

方案1：仅生成多个part文件（不按字段分区）

把coalesce(5)替换成repartition(5)，并且赋值给原DataFrame，强制重新洗牌数据生成指定数量的分区：

# 替换原来的mydf.coalesce(5)
mydf = mydf.repartition(5)

方案2：测试按字段分区（生成分区目录+多part文件）

如果你的“分区功能”是指按字段划分存储目录（比如按field3分区），可以加上partitionBy()方法，这样会生成对应字段值的子目录，每个目录下再按DataFrame分区数生成part文件：

# 先调整DataFrame分区数，再按字段分区写入
mydf = mydf.repartition(5)
mydf.write.partitionBy("field3").csv(path="s3a://bucket/output", header="true")

修正后的完整代码 #

from pyspark.sql.types import LongType, StringType, StructField, StructType, BooleanType, ArrayType, IntegerType, TimestampType
from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.ui.enabled", "true") \
    .config("spark.default.parallelism", "4") \
    .master("yarn-client") \
    .getOrCreate()

myschema = StructType([
    StructField("field1", TimestampType(), True),
    StructField("field2", TimestampType(), True),
    StructField("field3", StringType(), True),
    StructField("field4", StringType(), True),
    StructField("field5", StringType(), True)
])

# 修正了schema参数的笔误（原代码写的scheduled_schema应该是myschema）
mydf = spark.read.load("s3a://bucket/myfile.csv",
    format="csv",
    sep=",",
    timestampFormat="MM/dd/yyyy HH:mm:ss",
    header="true",
    schema=myschema
)

# 调整分区数，生成5个分区
mydf = mydf.repartition(5)

# 写入S3，可选加上partitionBy测试字段分区
mydf.write.csv(path="s3a://bucket/output",
    header="true"
)

额外注意点 #

• repartition vs coalesce：repartition会强制洗牌数据，能任意设置分区数（适合增加分区）；coalesce尽量不洗牌，只能安全减少分区，不适合你的场景。
• 懒执行规则：Spark的转换操作（比如repartition）不会立即执行，只有遇到write这类动作操作才会触发，所以必须把转换结果赋值给变量才能生效。

内容的提问来源于stack exchange，提问作者simplycoding