Spark中如何判定同一列的两个聚合操作是否等价？

你编写的测试用例中，直接比较同一列的col对象断言成功，但比较同一列的sum聚合Column对象、以及执行sum聚合后的DataFrame时断言均失败，核心原因是Spark对聚合类Column对象、DataFrame的默认equals实现并非基于逻辑语义，而是基于对象内部的唯一标识或执行计划的实例特征。

测试代码 #

package example

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}

class HelloSpec extends munit.FunSuite {
    test("compare identical columns") {
        val a = col("salary")
        val b = col("salary")
        // this works
        assertEquals(a, b)
    }
    test("compare identical aggregated columns") {
        val a = sum(col("salary"))
        val b = sum(col("salary"))
        // this fails
        assertEquals(a, b)
    }
    test("compare identical aggregated columns with data") {
        val spark = SparkSession.builder.appName("HelloSpec").master("local[*]").getOrCreate
        val schema = StructType(Array(
            StructField("name", StringType, true),
            StructField("salary", DoubleType, true)
        ))
        val data = Seq(
            Row("John", 50000.0),
            Row("Peter", 45000.0),
            Row("Tom", 47000.0)
        )
        val rdd = spark.sparkContext.parallelize(data)
        val df = spark.createDataFrame(rdd, schema)
        val a = df.select(sum(col("salary")))
        val b = df.select(sum(col("salary")))
        // this also fails
        assertEquals(a, b)
    }
}

一、比较聚合Column对象的逻辑等价性 #

Spark的聚合函数（如sum）生成的Column对象，默认equals会比较对象的实例ID或内部生成的唯一表达式ID，而非逻辑语义。要判断两个聚合Column是否语义相同，可以：

• 调用Column的expr方法获取字符串形式的表达式，直接比对内容

修改后的测试示例：

test("compare identical aggregated columns") {
    val a = sum(col("salary"))
    val b = sum(col("salary"))
    // 比较表达式的结构化字符串
    assertEquals(a.expr, b.expr)
}

二、比较聚合后的DataFrame的逻辑等价性 #

直接比较DataFrame对象的equals同样会失败，因为每个DataFrame都有独立的执行计划实例和唯一ID。验证语义一致性可从以下维度入手：

1. 比对Schema：确认字段名称、类型完全一致
2. 比对数据内容：将DataFrame转换为本地集合后比对（适合小数据量场景）
3. 比对执行计划：通过queryExecution.toString获取执行计划的结构化描述进行比对

修改后的测试示例：

test("compare identical aggregated columns with data") {
    val spark = SparkSession.builder.appName("HelloSpec").master("local[*]").getOrCreate
    val schema = StructType(Array(
        StructField("name", StringType, true),
        StructField("salary", DoubleType, true)
    ))
    val data = Seq(
        Row("John", 50000.0),
        Row("Peter", 45000.0),
        Row("Tom", 47000.0)
    )
    val rdd = spark.sparkContext.parallelize(data)
    val df = spark.createDataFrame(rdd, schema)
    val a = df.select(sum(col("salary")))
    val b = df.select(sum(col("salary")))
    
    // 1. 校验Schema一致性
    assertEquals(a.schema, b.schema)
    // 2. 校验数据内容一致性
    assertEquals(a.collect(), b.collect())
    // 3. 可选：校验执行计划一致性
    assertEquals(a.queryExecution.toString, b.queryExecution.toString)
}

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