Spark DataFrame列转行归一化及Databricks内存问题解决求助

核心实现逻辑 #

你的需求本质是宽表转长表（Unpivot），将同一演员的多部电影从横向列转为纵向行。针对超1000列的场景，必须避免硬编码列名，而是通过动态识别列规则生成转换逻辑：

1. 提取所有电影相关列的编号（如Movie_Name1中的1），将同编号的字段归为一组；
2. 动态构造stack函数表达式（Spark原生高效Unpivot算子），一次性完成宽转长，避免生成大量中间列。

内存优化关键措施 #

针对Databricks出现的GC失败、Photon内存超限、CodeCache满等问题，从代码逻辑和集群配置两方面优化：

• 代码层面：避免逐列操作生成中间DataFrame，直接用selectExpr执行动态生成的Unpivot逻辑，减少Driver和Executor的内存占用；
• 集群配置：
  • 调整CodeCache大小：spark.sql.codeCache.size=2g（根据集群规模调整，默认值可能不足）；
  • 优化GC参数：spark.driver.extraJavaOptions="-XX:ReservedCodeCacheSize=1g -XX:+UseG1GC"、spark.executor.extraJavaOptions="-XX:ReservedCodeCacheSize=1g -XX:+UseG1GC"；
  • 调整Photon缓冲区：spark.databricks.photon.bufferPool.size=2g（需确保已开启Photon：spark.databricks.photon.enabled=true）；
  • 开启Arrow优化：spark.sql.execution.arrow.pyspark.enabled=true，加速Python与JVM的数据传输。

完整Python代码示例 #

from pyspark.sql.functions import expr

# 假设输入DataFrame为df（Databricks中可直接使用已加载的df）
# df = spark.read.table("your_input_table")

# 1. 提取所有电影相关列，按编号分组
actor_col = "Actor"
movie_cols = [col for col in df.columns if col != actor_col]
movie_groups = {}
for col in movie_cols:
    # 从列名末尾提取数字，适配不同命名格式（如Movie_Budget_1）
    num = ''.join([c for c in col if c.isdigit()])
    if num not in movie_groups:
        movie_groups[num] = []
    movie_groups[num].append(col)

# 2. 动态构造stack表达式
stack_items = []
for movie_id, cols in movie_groups.items():
    # 按目标Schema匹配对应列
    name_col = next(c for c in cols if "Movie_Name" in c)
    director_col = next(c for c in cols if "Movie_Director" in c)
    budget_col = next(c for c in cols if "Movie_Budget" in c)
    week1_col = next(c for c in cols if "Movie_Collection_week1" in c)
    week2_col = next(c for c in cols if "Movie_collection_week2" in c)
    
    stack_items.extend([
        f"'{movie_id}' as Movie_ID",
        f"`{name_col}` as Movie_Name",
        f"`{director_col}` as Movie_Director",
        f"`{budget_col}` as Movie_Budget",
        f"`{week1_col}` as Movie_Collection_week1",
        f"`{week2_col}` as Movie_collection_week2"
    ])

stack_expr = f"stack({len(movie_groups)}, {', '.join(stack_items)})"

# 3. 执行Unpivot，过滤空电影记录
result_df = df.selectExpr(actor_col, stack_expr).filter("Movie_Name is not null")

# 验证结果Schema
result_df.printSchema()

补充说明 #

• 若列名编号规则不一致（如部分列用_N、部分用N），可调整编号提取逻辑适配实际格式；
• 数据量极大时，可先对输入DataFrame按Actor分区，减少每个Task处理的数据量；
• 优先使用原生stack算子，避免第三方melt函数或自定义UDF，性能更适配超宽表场景。

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