如何在Scala中仅追加Event Hub新增数据至Delta表（非流处理）

核心思路是手动维护Event Hub的消费偏移量，用一个Delta表充当checkpoint存储每次读取的结束位置，每次批处理时从这个位置开始读取新数据，读完后更新偏移量，以此实现只处理新增数据。

步骤1：创建存储偏移量的Checkpoint表 #

先建一个专门存消费位置的Delta表，用来记录每次读取Event Hub后的最新偏移量（仅第一次运行时执行）：

import org.apache.spark.sql.types._

val checkpointSchema = StructType(Seq(
  StructField("partitionId", StringType, nullable = false),
  StructField("offset", StringType, nullable = false),
  StructField("sequenceNumber", LongType, nullable = false),
  StructField("enqueuedTime", TimestampType, nullable = false)
))

// 初始化空的checkpoint表
spark.createDataFrame(spark.sparkContext.emptyRDD[Row], checkpointSchema)
  .write
  .format("delta")
  .mode("overwrite")
  .save("/path/to/checkpoint-delta-table")

步骤2：每次批处理的核心逻辑 #

每次运行批任务时，先读取checkpoint表获取上次的消费位置，以此作为本次读取Event Hub的起点；处理完数据后，把本次的最新偏移量更新回checkpoint表：

import org.apache.spark.sql.functions._

// 1. 读取checkpoint表，获取上次消费位置
val checkpointDF = spark.read.format("delta").load("/path/to/checkpoint-delta-table")

// 2. 配置Event Hub读取参数
val baseEventhubParams = Map(
  "eventhubs.namespace" -> "你的命名空间",
  "eventhubs.name" -> "你的Event Hub名称",
  "eventhubs.sas.key" -> "你的SAS密钥",
  "eventhubs.sas.key.name" -> "你的SAS密钥名称"
)

val finalEventhubParams = if (checkpointDF.isEmpty) {
  // 第一次运行，使用你原来的起始时间戳配置
  baseEventhubParams + ("eventhubs.startingPosition" -> """{"enqueuedTime":"2024-01-01T00:00:00Z","isInclusive":true}""")
} else {
  // 非首次运行，按分区生成起始位置JSON
  val startingPositions = checkpointDF
    .select("partitionId", "offset")
    .rdd
    .map(row => (row.getString(0), row.getString(1)))
    .collectAsMap()
    .map { case (partition, offset) =>
      s""""$partition": {"offset":"$offset","isInclusive":false}"""
    }
    .mkString("{", ",", "}")
  
  baseEventhubParams + ("eventhubs.startingPosition" -> startingPositions)
}

// 3. 读取Event Hub数据
val ehDF = spark.read
  .format("eventhubs")
  .options(finalEventhubParams)
  .load()

// 4. 执行你的数据转换逻辑（关联、选择等）
// 先提取body中的业务数据，再做自定义转换
val transformedDF = ehDF
  .select(
    from_json(col("body").cast(StringType), 你的业务数据Schema).alias("data"),
    col("partitionId"),
    col("offset"),
    col("sequenceNumber"),
    col("enqueuedTime")
  )
  .select("data.*", "partitionId", "offset", "sequenceNumber", "enqueuedTime")
  // 这里添加你的关联、过滤、选择操作
  .join(其他表, Seq("关联字段"), "left")
  .select("需要保留的字段", "partitionId", "offset", "sequenceNumber", "enqueuedTime")

// 5. 将转换后的数据追加到目标Delta表
transformedDF
  .drop("partitionId", "offset", "sequenceNumber", "enqueuedTime") // 移除不需要的元数据字段
  .write
  .format("delta")
  .mode("append")
  .save("/path/to/target-delta-table")

// 6. 更新checkpoint表：按分区保留最新的消费位置
val latestOffsetsDF = ehDF
  .groupBy("partitionId")
  .agg(max("sequenceNumber").alias("maxSeqNo"))
  .join(ehDF, Seq("partitionId", "sequenceNumber"), "inner")
  .select("partitionId", "offset", "sequenceNumber", "enqueuedTime")

latestOffsetsDF
  .write
  .format("delta")
  .mode("overwrite")
  .save("/path/to/checkpoint-delta-table")

关键说明 #

• 偏移量准确性：通过按分区取最大的sequenceNumber来锁定每个分区的最新消费位置，确保下次读取不会重复处理已有数据。
• 调度要求：这种方式是批处理模式，需要借助调度工具（如Airflow、Azure Data Factory）定期运行，周期可根据你的数据推送频率调整。
• 失败恢复：如果批任务执行失败，checkpoint表不会更新，下次运行会从上次成功的位置重新读取，避免数据丢失或重复写入。

内容的提问来源于stack exchange，提问作者Tamás Godányi