如何用PySpark按首字母拆分XML并高效分区写入S3存储

DevOps 小满

2026-10-14

优化方案

原代码性能问题原因

循环遍历26个字母执行过滤+写入操作，每次写入都是独立的Spark Action，会触发全量数据的扫描、计算共26次，IO和计算资源消耗是单次任务的26倍，超大文件场景下耗时会非常高。

最优实现方案

核心逻辑是仅扫描全量数据1次，通过分区写入完成分类，后续补建空文件夹满足所有首字母路径存在的要求。

步骤1：新增首字母分区列，一次写入所有分类

import pyspark.sql.functions as F

# 新增首字母分区列
df_with_partition = df.withColumn(
    "first_char",
    F.lower(F.trim(F.col("name")).substr(1, 1))
)

# 按首字母分区一次写入，仅触发1次全量扫描
df_with_partition.write\
    .format("com.databricks.spark.xml")\
    .option("maxRecordsPerFile", 800)\
    .option("rootTag","source")\
    .option("rowTag", "employees")\
    .mode("overwrite")\
    .partitionBy("first_char")\
    .save("s3://split-files/")

如果不需要路径带first_char=前缀，Spark 3.2+版本可以开启配置去掉分区列前缀，直接生成s3://split-files/a/格式的路径：

spark.conf.set("spark.sql.sources.partitionColumnNamingStrategy", "none")

步骤2：补建无数据的首字母空文件夹

S3的文件夹为虚拟前缀，仅需对无数据的首字母路径写入一个空文件即可生成对应路径，操作仅需毫秒级：

import boto3

s3_client = boto3.client("s3")
bucket = "split-files"
all_chars = [chr(ord("a") + i) for i in range(26)]

for char in all_chars:
    # 检查当前首字母路径是否存在数据
    resp = s3_client.list_objects_v2(
        Bucket=bucket,
        Prefix=f"{char}/",
        MaxKeys=1
    )
    # 无数据则写入空标记文件生成路径
    if "Contents" not in resp:
        s3_client.put_object(
            Bucket=bucket,
            Key=f"{char}/.empty",
            Body=b""
        )