Spark Streaming用commitAsync提交偏移时，处理失败如何不提交对应Topic A偏移？

Great question! Let's walk through exactly how to implement this requirement, focusing on per-record failure handling and precise offset management since you want to avoid committing offsets for failed records.

Core Approach #

Since you need granular control over which offsets get committed (only those corresponding to successfully processed records), we'll use manual offset management with Spark Streaming's Direct Kafka Stream. The fact that Topic A and Topic B have the same number of partitions (N) is perfect—it lets us map partitions one-to-one, preserving order and simplifying offset tracking.

Step-by-Step Implementation #

1. Base Configuration & Stream Initialization

First, set up your Spark Streaming context and Kafka parameters, making sure to disable auto-commit (we'll handle this manually):

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

// Initialize Spark Streaming context (adjust batch interval as needed)
val sparkConf = new SparkConf().setAppName("TopicAToTopicBStream")
val ssc = new StreamingContext(sparkConf, Seconds(5))

// Kafka consumer configuration
val kafkaConsumerParams = Map[String, Object](
  "bootstrap.servers" -> "your-kafka-brokers:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "spark-topic-a-consumer-group",
  "enable.auto.commit" -> (false: java.lang.Boolean), // Disable auto-commit
  "auto.offset.reset" -> "latest" // Adjust based on your needs (e.g., "earliest")
)

// Create Direct Stream (required for access to offset ranges)
val topics = Array("TopicA")
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topics, kafkaConsumerParams)
)

2. Per-Partition Processing & Offset Tracking

We'll process each partition individually (leveraging the 1:1 partition mapping between Topic A and B), track successful records, and only commit offsets up to the last successfully processed record.

First, implement a singleton Kafka Producer to avoid creating a new instance for every partition (critical for performance):

import java.util.Properties

object KafkaProducerSingleton {
  private var producer: KafkaProducer[String, String] = null

  def getInstance(): KafkaProducer[String, String] = {
    if (producer == null) {
      val props = new Properties()
      props.put("bootstrap.servers", "your-kafka-brokers:9092")
      props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      props.put("acks", "all") // Ensure message is fully persisted before proceeding
      props.put("enable.idempotence", "true") // Optional: For exactly-once delivery
      producer = new KafkaProducer[String, String](props)
    }
    producer
  }
}

Now add the processing logic to your stream:

stream.foreachRDD { rdd =>
  // Get offset ranges for the current RDD (only available with Direct Stream)
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

  // Process each Kafka partition separately
  rdd.foreachPartition { partitionRecords =>
    val taskPartitionId = TaskContext.get.partitionId
    val currentOffsetRange = offsetRanges(taskPartitionId)
    val topicPartition = new TopicPartition(currentOffsetRange.topic, currentOffsetRange.partition)

    val producer = KafkaProducerSingleton.getInstance()
    var lastSuccessfulOffset = currentOffsetRange.fromOffset - 1L // Start at position before first record

    try {
      partitionRecords.foreach { record =>
        // Map Topic A record directly to Topic B's matching partition
        val topicBRecord = new ProducerRecord[String, String](
          "TopicB",
          record.partition(), // Match Topic A partition to Topic B partition
          record.key(),
          record.value() // Replace with your transformed value if needed
        )

        // Synchronous send to confirm success before moving on
        val metadata: RecordMetadata = producer.send(topicBRecord).get()
        
        // Update last successful offset only if send succeeds
        lastSuccessfulOffset = record.offset()
      }

      // Commit offsets if we processed at least one record successfully
      if (lastSuccessfulOffset >= currentOffsetRange.fromOffset) {
        val offsetCommitMap = Map(topicPartition -> (lastSuccessfulOffset + 1L))
        // Commit async to avoid blocking (use commitSync for strict consistency)
        KafkaUtils.commitAsync(kafkaConsumerParams, offsetCommitMap)
      }
    } catch {
      case e: Exception =>
        // If any record fails, commit up to the last successful offset (if any)
        if (lastSuccessfulOffset >= currentOffsetRange.fromOffset) {
          val offsetCommitMap = Map(topicPartition -> (lastSuccessfulOffset + 1L))
          KafkaUtils.commitAsync(kafkaConsumerParams, offsetCommitMap)
        }
        // Add logging/alerts here for failed records
        println(s"Failed processing partition ${currentOffsetRange.partition}. Last successful offset: $lastSuccessfulOffset. Error: ${e.getMessage}")
    }
  }
}

// Start the stream and await termination
ssc.start()
ssc.awaitTermination()

Key Details to Note #

• Direct Stream Requirement: We use createDirectStream because it gives us direct access to OffsetRange objects, which are essential for manual offset management.
• 1:1 Partition Mapping: Since Topic A and B have the same number of partitions, we send records from Topic A's partition i directly to Topic B's partition i. This preserves order within partitions and simplifies offset tracking.
• Offset Commit Logic: Kafka expects you to commit the offset of the next record to consume, so we add 1 to the last successful offset.
• Failure Handling: If a record fails to send, we only commit offsets up to the last successful record. The next batch will reprocess from the failed record's position, ensuring no data is lost.
• Idempotency: For exactly-once delivery, enable producer idempotence (enable.idempotence=true) and ensure your processing logic is idempotent (e.g., use unique record IDs to avoid duplicate effects).

Optional Enhancements #

• Retry Logic: Add retries for transient failures (e.g., network blips) using a loop with backoff before failing the record.
• Checkpointing: Enable Spark Streaming checkpointing to persist offset information across restarts, ensuring you resume from the last committed offset.
• Monitoring: Track offset lag and failed record counts using tools like Prometheus or Grafana to catch issues early.

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