Spark 2.4迁移至3.3（Azure Synapse）后出现StackOverflowError

问题原因 #

你遇到的java.lang.StackOverflowError是因为大量链式调用withColumn导致Spark逻辑计划的树结构嵌套过深。Spark 3.x版本的Catalyst优化器在逻辑计划的分析、克隆阶段，对递归遍历树结构的处理比2.4版本更严格。当DataFrame经过数十甚至上百次withColumn操作后，逻辑树的深度会急剧增加，触发栈溢出错误。

解决方案 #

1. 合并窗口函数到单个select操作 #

将所有窗口函数计算合并到同一个select中，替代多次链式调用withColumn，避免生成嵌套过深的Project节点。

修改后示例代码：

def dataframeAvg(spark: SparkSession, df: DataFrame): DataFrame = {
    import spark.implicits._

    val dfAvg = df.select(
        // 保留原表所有列
        col("*"),
        // 窗口计算列一次性定义
        row_number().over(windowTrans).alias("row"),
        count("*").over(windowTrans).alias("LONG_HIST"),
        
        avg("col1").over(windowTrans).alias("AVG_COL1"),
        avg("col2").over(windowTrans).alias("AVG2"),
        avg("col3").over(windowTrans).alias("AVG3"),
        avg("col6").over(windowTrans).alias("AVG6"),
        
        stddev_samp("col1").over(windowTrans).alias("STD_COL1"),
        stddev_samp("col2").over(windowTrans).alias("STD2"),
        stddev_samp("col3").over(windowTrans).alias("STD3"),
        stddev_samp("col6").over(windowTrans).alias("STD6"),
        
        min("col1").over(windowTransMinMax).alias("MIN_COL1"),
        min("col2").over(windowTransMinMax).alias("MIN2"),
        min("col3").over(windowTransMinMax).alias("MIN3"),
        min("col6").over(windowTransMinMax).alias("MIN6")
    )
    .where(col("row") === 1)
    .select(/* 目标列列表 */)
    
    dfAvg
}

2. 拆分计算阶段并缓存截断逻辑树 #

如果无法一次性合并所有计算，可将操作拆分为多个阶段，每个阶段完成部分列计算后调用cache()或persist()，截断逻辑树的嵌套深度。

示例：

def dataframeAvg(spark: SparkSession, df: DataFrame): DataFrame = {
    import spark.implicits._

    // 第一阶段：计算row、LONG_HIST及平均值/标准差列
    val step1 = df.withColumn("row", row_number().over(windowTrans))
        .withColumn("LONG_HIST", count("row").over(windowTrans))
        .withColumn("AVG_COL1", avg("col1").over(windowTrans))
        .withColumn("AVG2", avg("col2").over(windowTrans))
        .withColumn("STD_COL1", stddev_samp("col1").over(windowTrans))
        .withColumn("STD2", stddev_samp("col2").over(windowTrans))
        .cache() // 缓存后截断逻辑树

    // 第二阶段：计算极值列并完成过滤、选择
    val dfAvg = step1.withColumn("MIN_COL1", min("col1").over(windowTransMinMax))
        .withColumn("MIN2", min("col2").over(windowTransMinMax))
        .where(col("row") === 1)
        .select(/* 目标列列表 */)
    
    // 释放缓存
    step1.unpersist()
    dfAvg
}

3. 优化窗口定义 #

windowTrans的orderBy中重复了partitionBy的列，这属于冗余定义（分区内这些列值完全一致，排序无意义），可简化窗口定义，减少不必要的计算压力：

def windowTrans = Window.partitionBy("COD_X", "COD_Y", "date")
    // 若不需要特定排序，可去掉orderBy；若需要，替换为实际需要排序的列
    .orderBy(/* 必要的排序列，如时间戳等 */)

补充说明 #

Spark 3.x对Catalyst优化器的逻辑计划处理逻辑进行了升级，递归遍历的深度限制更严格。而Spark 2.4版本对嵌套较深的逻辑树兼容性更好，因此相同代码在旧版本可正常运行。上述方案通过减少逻辑树的嵌套深度，从根本上解决栈溢出问题。

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