Apache Spark Structured Streaming混合数据源容错实现方案咨询

需求背景 #

需要开发一个Spark Structured Streaming应用，先处理HDFS上Avro格式的Kafka归档历史数据，完成后无缝切换到Kafka实时主题处理，目标是实现类似Apache Flink混合源的功能（该组件支持先处理批数据源，完成后自动切换到流数据源，实现批流一体化处理），同时保障全流程的容错性。

当前初步方案：

• 读取Avro文件：使用spark.read.format("avro").load("hdfs://...")加载历史数据
• 切换至Kafka流处理：通过readStream.format("kafka")...配置实时流处理

以下是针对该场景的容错设计实现方案：

核心容错设计方案 #

1. 统一数据模型与处理逻辑复用 #

为批、流阶段定义统一的Schema和处理函数，避免逻辑不一致，同时降低维护成本：

// 定义业务数据统一Schema
val eventSchema = new StructType()
  .add("eventId", StringType)
  .add("eventTime", TimestampType)
  .add("payload", StringType)

// 通用数据处理函数
def processEvent(df: DataFrame): DataFrame = {
  df.select(
    col("eventId"),
    col("eventTime"),
    from_json(col("payload"), eventSchema).alias("biz_data")
  )
  // 后续业务转换、聚合等逻辑...
}

2. 批处理阶段的容错保障 #

处理Avro历史数据时，确保Exactly-Once语义，避免重复处理或数据丢失：

• 幂等输出：如果输出到数据湖（如Delta Lake、Hudi），利用其ACID特性或按时间分区覆盖未完成的分区；如果输出到关系型数据库，通过主键UPSERT或去重逻辑实现幂等写入。
• 进度持久化：将已处理的Avro文件路径/最大事件时间，原子写入分布式存储（如HDFS、ZooKeeper），作为后续切换到流处理的触发标记，同时用于故障恢复时跳过已处理数据。

3. 流处理阶段的容错配置 #

切换到Kafka实时流时，开启Spark Structured Streaming的端到端Exactly-Once语义：

• 配置Checkpoint：指定分布式存储上的Checkpoint目录，用于恢复作业状态和Kafka消费offset：

// 从批处理进度中获取Kafka起始offset
def getKafkaStartingOffset(): String = {
  // 读取之前持久化的批处理最大eventTime，调用Kafka offsetsForTimes API获取对应offset
  "{\"topic_name\":{0:12345}}"
}

val kafkaStreamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "topic_name")
  .option("startingOffsets", getKafkaStartingOffset())
  .option("enable.auto.commit", "false") // 由Spark管理offset提交
  .load()
  .select(col("value").cast(StringType).alias("payload"))
  .transform(processEvent)
  .writeStream
  .option("checkpointLocation", "hdfs:///spark/checkpoint/kafka_stream")
  .format("delta") // 选择支持事务的输出源
  .start()

• 输出源选型：优先选择支持事务的存储（如Delta Lake、Kafka Sink），保障端到端Exactly-Once。

4. 无缝切换的容错触发机制 #

实现批处理完成后自动切换到流处理，且保证数据无重叠、无丢失：

• 作业监控触发：提交批处理作业后，监听作业状态；当作业成功完成时，读取批处理进度记录，通过Kafka offsetsForTimes API获取对应起始offset，启动流处理作业。
• 原子性校验：启动流处理前，校验批处理进度与Kafka起始offset的一致性，避免因时间戳与offset映射误差导致的数据遗漏。
• 避免重复触发：在分布式存储中写入流处理启动标记，重启时先检查标记，避免重复启动流作业。

5. 异常恢复机制 #

• 批处理故障恢复：重新运行批处理时，读取已处理文件列表/进度记录，只处理未完成的Avro文件，跳过已处理数据。
• 流处理故障恢复：利用Spark Checkpoint机制，重启作业时自动恢复到故障前的状态，继续消费未处理的Kafka数据。
• 切换环节故障恢复：如果切换过程中出现故障，重新触发时先验证批处理是否完成、流处理是否已启动，确保状态一致后再执行切换。

简化方案：流读批数据+流合并 #

利用Spark Structured Streaming支持将批数据作为流读取的特性，合并批、流数据源，实现自动切换：

// 读取Avro文件作为流（一次触发处理所有历史数据）
val batchStreamDF = spark.readStream
  .format("avro")
  .schema(eventSchema)
  .load("hdfs:///path/to/avro_archive")
  .withColumn("source_type", lit("batch"))

// 读取Kafka实时流
val kafkaStreamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("subscribe", "topic_name")
  .option("startingOffsets", "latest")
  .load()
  .select(from_json(col("value").cast(StringType), eventSchema).alias("data"))
  .withColumn("source_type", lit("stream"))

// 合并两个流，按eventTime排序保证先处理历史数据
val combinedDF = batchStreamDF.union(kafkaStreamDF)
  .orderBy(col("eventTime"))
  .transform(processEvent)
  .writeStream
  .option("checkpointLocation", "hdfs:///spark/checkpoint/combined_stream")
  .format("delta")
  .start()

该方案依赖Spark流的Checkpoint实现容错，无需手动管理批流切换，但需注意通过eventTime排序保障处理顺序。

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