从Kafka读取索引名后执行Spark任务遇重复读及执行失败问题

问题1：Kafka消费者循环poll重复读取消息 #

原因 #

你的循环代码未提交消费偏移量。即便配置了相同Group ID，Kafka需要确认消息已被处理才会更新该Group对应的偏移量位置。如果不提交偏移量，每次poll时Kafka会判定消息未处理，重复返回。

解决方案 #

处理完每批消息后手动提交偏移量，修改代码如下：

while (true) {
    val records = kafkaClient.poll(Duration.ofMillis(1000))
    if (!records.isEmpty) {
        records.forEach { record ->
            val index = record.value()
            // 执行读取Elasticsearch索引的代码
        }
        // 同步提交偏移量，确保提交成功再继续后续poll
        kafkaClient.commitSync()
    }
}

若想避免阻塞，也可使用异步提交commitAsync()，但同步提交更能保证消息不重复处理。另外也可开启自动提交（需在创建消费者时配置），但自动提交存在“提交偏移量后消息处理失败”的丢消息风险，不推荐在数据可靠性要求高的场景使用：

// 创建Kafka消费者时添加配置
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")

问题2：Structured Streaming中执行ES读取抛出Writing job aborted异常 #

原因 #

你在map算子内直接调用SparkSession读取ES，存在两个核心问题：

1. map是运行在Executor节点的分布式算子，而SparkSession是Driver端创建的对象，无法序列化传递到Executor中复用；
2. 分布式算子内部嵌套Spark读写操作，会打乱任务调度逻辑，触发作业提交异常。

解决方案 #

使用Structured Streaming的foreachBatch算子，它允许在Driver端针对每个微批处理数据，可安全执行ES读取任务。修改代码如下：

val df = session.readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "test_consumers")
    .option("startingOffsets", "earliest")
    .load()

val indexDf = df.selectExpr("CAST(value AS STRING) AS index_name")

// 用foreachBatch处理每个微批的索引名
val query = indexDf.writeStream
    .foreachBatch { (batchDf: DataFrame, batchId: Long) =>
        // 收集当前微批的所有索引名（适用于索引数量不多的场景）
        val indexes = batchDf.select("index_name").as[String].collect()
        indexes.foreach { index =>
            // 执行读取Elasticsearch索引的代码
            val esDataset: Dataset[Row] = session.read()
                .format("org.elasticsearch.spark.sql")
                .option("es.read.field.include", "orgUUID,serializedEventKey,involvedContactURNs,crmAssociationSmartURNs")
                .option("es.read.field.as.array.include", "involvedContactURNs,crmAssociationSmartURNs")
                .load(index)
            esDataset.foreach(transform)
        }
    }
    .option("checkpointLocation", "/path/to/your/checkpoint") // 必须设置checkpoint路径
    .start()

query.awaitTermination()

注意事项：

• foreachBatch必须配置checkpointLocation，否则流任务无法启动；
• 若微批中索引数量极大，collect()拉取到Driver端处理会有性能瓶颈，可改为将索引名广播到Executor，用mapPartitions分布式处理ES读取逻辑。

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