基于event.event_name拆分GZIP JSON并转CSV至S3的性能优化

针对你当前5小时的处理耗时，我们从读取效率、扁平化复用、分区存储三个核心环节优化，以下是具体可落地的方案：

一、优化单行GZIP JSON读取 #

你的文件是包含500万条记录的单行JSON数组（[{"event":...}, {...}]），Spark默认JSON Reader会将整个文件视为单条记录，导致解析极慢。改用textRDD读取后拆分数组的方式，最大化并行处理：

# 读取GZIP文本（原耗时4秒，保留该高效步骤）
text_rdd = spark.sparkContext.textFile("s3://your-input-path/file.gz")

# 拆分单行数组为独立JSON对象，并行处理
split_rdd = text_rdd.flatMap(
    lambda x: [f"{{{item}}}" for item in x.strip()[1:-1].split("}, {")]
)

# 可选：用样本数据预推断Schema，避免全量扫描推断的耗时
sample_json = split_rdd.takeSample(False, 1000)
sample_df = spark.read.json(spark.sparkContext.parallelize(sample_json))
final_schema = sample_df.schema

# 用预推断的Schema读取全量数据，大幅提升转DF效率
raw_df = spark.read.json(split_rdd, schema=final_schema)

此步骤可将原textRDD转DF的300秒耗时压缩至数十秒内。

二、扁平化JSON：只做一次，全局复用 #

你当前每个event_name重复执行扁平化+字段选择，属于重复计算。改为一次性扁平化全量数据，后续直接使用扁平结构：

from pyspark.sql.types import StructType
from pyspark.sql.functions import col

def flatten_df(df):
    def _flatten_schema(schema, prefix=""):
        flat_cols = []
        for field in schema.fields:
            field_path = f"{prefix}{field.name}"
            if isinstance(field.dataType, StructType):
                flat_cols.extend(_flatten_schema(field.dataType, f"{field_path}."))
            else:
                # 将嵌套路径转为下划线命名，如event.event_name → event_event_name
                flat_cols.append(col(field_path).alias(field_path.replace(".", "_")))
        return flat_cols
    
    return df.select(_flatten_schema(df.schema))

# 仅执行一次扁平化（原耗时180秒，保留但只做一次）
flat_df = flatten_df(raw_df)

# 提取event_name作为后续分区字段
flat_df = flat_df.withColumn("event_name", col("event_event_name"))

三、按event_name分区存储：避免70次重复过滤 #

放弃逐个event_name过滤保存的方式，改用Spark的partitionBy一次性完成分区写入，Spark会自动按event_name生成对应文件夹，每个文件夹下存储该类型的CSV文件：

# 调整Shuffle并行度，适配2个DPU的8核配置（可根据实际情况微调）
spark.sql("SET spark.sql.shuffle.partitions=16")

# 分区写入S3，自动按event_name拆分文件
flat_df.write \
    .mode("overwrite") \
    .partitionBy("event_name") \
    .option("header", "true") \
    .option("emptyValue", "")  # 将Null转为空字符串适配CSV格式
    .csv("s3://your-output-path/")

此步骤替代原70次单独过滤+保存的操作，将原本7000秒的耗时压缩至数百秒内。

四、额外性能提升建议 #

1. 增加DPU数量：当前2个DPU的资源对于500万条动态JSON处理较为紧张，建议临时扩容至4-8个DPU，并行处理能力可翻倍提升。
2. 缓存中间结果：若内存充足（2个DPU共32GB），可在扁平化后执行flat_df.cache()，避免写入阶段重复计算扁平结构。
3. 预处理拆分大文件：如果该类单行GZIP文件是周期性生成的，可提前用AWS Lambda将其拆分为多个小JSON文件（每行一条记录），Spark读取时可并行处理多个文件，进一步降低读取耗时。

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