替代DataFrame.explode()的高性能方案：大列表列数据展开

当处理超大规模包含列表类型列的DataFrame时，explode()容易出现OOM或性能瓶颈，以下是几种更高效的替代方案：

方案1：RDD flatMap 实现（灵活且低开销） #

利用RDD的底层flatMap操作直接拆分列表，避免DataFrame API可能带来的额外优化开销，步骤如下：

仅拆分目标列表列（无其他关联列） #

# 将DataFrame转为RDD
rdd = df.rdd
# flatMap拆分每个列表元素为独立Row
exploded_rdd = rdd.flatMap(lambda row: [(cid,) for cid in row.category_id])
# 转回DataFrame，显式指定schema以避免自动推断的性能损耗
from pyspark.sql.types import StringType, StructType, StructField
schema = StructType([StructField("category_id", StringType(), nullable=True)])
df_exploded = spark.createDataFrame(exploded_rdd, schema)

保留其他关联列（如销售日期等） #

如果原DataFrame包含需要保留的其他字段，只需在flatMap时携带对应数据：

exploded_rdd = rdd.flatMap(lambda row: [(row.sales_date, cid) for cid in row.category_id])
# 定义对应schema
schema = StructType([
    StructField("sales_date", StringType(), nullable=True),
    StructField("category_id", StringType(), nullable=True)
])
df_exploded = spark.createDataFrame(exploded_rdd, schema)

方案2：优化原生explode() （简单高效） #

若不想切换到RDD，可以通过调整Spark分区与资源配置解决OOM问题，同时提升性能：

1. 提前重分区：将原DataFrame拆分为更多小分区，避免单个分区数据量过载
2. 调整executor内存：根据集群资源增加executor内存分配（如启动时指定--executor-memory 8g）

示例代码：

# 重分区后执行explode，推荐分区数为集群核心数的2-3倍
df_exploded = df.repartition(200).explode("category_id")

方案3：Spark SQL posexplode() （需保留原列表索引场景） #

如果需要保留原列表中元素的位置信息，可以使用posexplode()，性能与优化后的explode()相当：

-- 注册临时视图
df.createOrReplaceTempView("sales_data")
-- 执行SQL拆分
SELECT pos, category_id FROM sales_data LATERAL VIEW posexplode(category_id) exploded AS pos, category_id

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