Spark 3.2.2结构化流Join结果与预期数据比对方案咨询

一、解决Schema对比返回false的问题 #

直接调用StructType.equals()会因为实例引用不同返回false，即使结构完全一致。需改用Spark提供的专用对比方法或手动校验：

• 宽松对比（忽略nullable属性）：使用actualSchema.equalsIgnoreNullability(expectedSchema)，该方法只校验字段名、数据类型及嵌套结构，忽略是否允许为null的差异，适合多数测试场景。
• 严格对比：若需要校验nullable属性，可递归遍历两个Schema的字段，逐个检查name、dataType、nullable三个属性：

  def schemaEqualsStrict(actual: StructType, expected: StructType): Boolean = {
    if (actual.fields.length != expected.fields.length) return false
    actual.fields.zip(expected.fields).forall { case (a, e) =>
      a.name == e.name && a.dataType == e.dataType && a.nullable == e.nullable &&
        (a.dataType match {
          case struct: StructType => schemaEqualsStrict(struct, e.dataType.asInstanceOf[StructType])
          case _ => true
        })
    }
  }
  

二、流输出结果的验证方法 #

1. foreachBatch内直接验证（推荐测试场景） #

针对测试数据量小的情况，在foreachBatch中直接将批次数据收集到Driver端，与预期数据对比：

// 预定义预期数据和Schema
val expectedRows = Seq(
  Row("id1", "name1", "value1"),
  Row("id2", "name2", "value2")
)
val expectedSchema = StructType(Seq(
  StructField("id", StringType),
  StructField("name", StringType),
  StructField("value", StringType)
))

// 执行Join并验证
val joinedStream = leftStream.join(rightStream, Seq("id"), "inner")
val query = joinedStream.writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    // 验证Schema
    assert(batchDF.schema.equalsIgnoreNullability(expectedSchema), "Schema不匹配")
    // 验证数据（先排序消除顺序影响）
    val actualRows = batchDF.orderBy("id").collect()
    val sortedExpected = expectedRows.sortBy(_.getAs[String]("id"))
    assert(actualRows sameElements sortedExpected, "数据内容不匹配")
  }
  .start()

// 等待处理完成（测试场景设置超时）
query.awaitTermination(30000)
query.stop()

2. 输出到文件后验证 #

若必须通过文件输出验证，需注意过滤临时文件并读取全量输出：

val outputPath = "/tmp/join-test-output"
val checkpointPath = "/tmp/join-test-checkpoint"

// 清理历史数据
new File(outputPath).delete()
new File(checkpointPath).delete()

// 启动流写入JSON文件
val query = joinedStream.writeStream
  .format("json")
  .option("path", outputPath)
  .option("checkpointLocation", checkpointPath)
  .outputMode("append")
  .start()

query.awaitTermination(30000)
query.stop()

// 读取输出文件（过滤Spark生成的临时文件，如._开头的文件）
val actualDF = spark.read.json(
  spark.sparkContext.wholeTextFiles(s"$outputPath/*")
    .filter(!_._1.contains("_"))
    .values
)

// 验证Schema和数据
assert(actualDF.schema.equalsIgnoreNullability(expectedSchema))
val actualRows = actualDF.orderBy("id").collect()
val expectedDF = spark.createDataFrame(expectedRows, expectedSchema)
val expectedRowsSorted = expectedDF.orderBy("id").collect()
assert(actualRows sameElements expectedRowsSorted)

三、单元测试方案（ScalaTest示例） #

使用本地SparkSession模拟流环境，结合内存数据源生成测试数据，完成自动化测试：

import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.scalatest.BeforeAndAfterAll
import org.scalatest.funsuite.AnyFunSuite

class StreamJoinUnitTest extends AnyFunSuite with BeforeAndAfterAll {
  private var spark: SparkSession = _

  override def beforeAll(): Unit = {
    spark = SparkSession.builder()
      .master("local[2]")
      .appName("StreamJoinTest")
      .getOrCreate()
  }

  override def afterAll(): Unit = {
    spark.stop()
  }

  test("内连接结果符合预期") {
    // 定义输入Schema和测试数据
    val leftSchema = StructType(Seq(StructField("id", StringType), StructField("name", StringType)))
    val rightSchema = StructType(Seq(StructField("id", StringType), StructField("value", StringType)))
    val leftData = Seq(Row("id1", "name1"), Row("id2", "name2"))
    val rightData = Seq(Row("id1", "value1"), Row("id2", "value2"))

    // 创建内存模拟流
    spark.createDataFrame(spark.sparkContext.parallelize(leftData), leftSchema)
      .writeStream.format("memory").queryName("leftStream").start()
    spark.createDataFrame(spark.sparkContext.parallelize(rightData), rightSchema)
      .writeStream.format("memory").queryName("rightStream").start()

    // 执行Join
    val leftDF = spark.readStream.table("leftStream")
    val rightDF = spark.readStream.table("rightStream")
    val joinedDF = leftDF.join(rightDF, Seq("id"), "inner")

    // 预定义预期结果
    val expectedSchema = StructType(Seq(
      StructField("id", StringType),
      StructField("name", StringType),
      StructField("value", StringType)
    ))
    val expectedRows = Seq(Row("id1", "name1", "value1"), Row("id2", "name2", "value2"))

    // 验证逻辑
    var testPassed = true
    val query = joinedDF.writeStream
      .foreachBatch { (batchDF: DataFrame, _: Long) =>
        if (!batchDF.schema.equalsIgnoreNullability(expectedSchema)) {
          testPassed = false
          println(s"Schema不匹配：实际=${batchDF.schema.json}，预期=${expectedSchema.json}")
        }
        val actualRows = batchDF.orderBy("id").collect()
        val sortedExpected = expectedRows.sortBy(_.getAs[String]("id"))
        if (!(actualRows sameElements sortedExpected)) {
          testPassed = false
          println(s"数据不匹配：实际=${actualRows.mkString(",")}，预期=${sortedExpected.mkString(",")}")
        }
      }
      .start()

    // 等待处理完成
    query.processAllAvailable()
    query.stop()
    spark.streams.active.foreach(_.stop())

    assert(testPassed, "流Join结果验证失败")
  }
}

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