如何在Spark Structured Streaming应用中计算Kafka lag

问题1：是否可以通过Spark接口编程获取指定Kafka主题所有分区的最新偏移量 #

可以，共有两种实现路径：

• 方式一：使用Spark Kafka连接器内置的偏移量读取工具
  Spark 2.3及以上版本的spark-sql-kafka-0-10连接器内置了KafkaOffsetReader类，传入和读流任务一致的Kafka连接配置（bootstrap.servers、SSL/SASL认证配置等）、目标主题列表，调用fetchLatestOffsets()方法就能直接返回指定主题所有分区的最新偏移量。
  注意：这个类属于连接器内部API，Spark跨大版本升级时经常调整内部类路径、方法签名，没有官方兼容性保证，生产环境强依赖的话要做好对应版本的适配。
  参考实现代码（Scala）：

import org.apache.spark.sql.kafka010.{KafkaOffsetReader, KafkaOffsetRangeStrategy, KafkaParams}
import org.apache.kafka.common.serialization.StringDeserializer

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "kafka-cluster:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "your-streaming-job"
)
val offsetReader = KafkaOffsetReader.create(
  KafkaOffsetRangeStrategy(Seq("target_topic")),
  KafkaParams(kafkaParams),
  spark.sessionState.newHadoopConf()
)
// 返回值即为目标主题所有分区的最新偏移量
val latestOffsets = offsetReader.fetchLatestOffsets()

• 方式二：通过流查询监听器获取偏移量快照
  可以给StreamingQuery注册自定义StreamingQueryListener，每次批次触发QueryProgress事件时，从事件的sources字段中提取Kafka源记录的偏移量快照。
  注意：这个快照是当前批次启动拉取数据瞬间的偏移量值，等你完成这批数据的业务处理时，Kafka已经写入了新消息，用这个值算出来的lag会比真实值偏小，只适合粗粒度监控场景，要精确lag值不建议用这个方案。

问题2：是否可以在Spark应用中直接调用Kafka原生Admin接口获取批次对应的最新偏移量 #

完全可以，而且这是生产环境最推荐的实现方案——Kafka官方提供的AdminClient有严格的跨版本兼容性保证，不会因为Spark版本升级出现代码不兼容的问题，稳定性比依赖Spark内部API高很多。
实现时注意几个核心要点即可：

• AdminClient实例要在Driver端初始化（比如main方法、foreachBatch的单例初始化逻辑中创建），不要在map/filter这类运行在executor端的算子逻辑里创建实例，避免每个executor都和Kafka建立长连接，打满broker端的连接数。AdminClient本身是线程安全的，整个应用维持一个全局实例就行。
• 每个批次计算lag时，先从你已经解析到的已处理偏移量（流数据中携带的Kafka元数据）整理成TopicPartition维度的已消费位置，再调用AdminClient.listOffsets()方法，传入OffsetSpec.latest()获取所有目标分区的实时最新偏移量，两个值做差就是对应分区的真实lag。
• 给应用加JVM关闭钩子，在流任务停止时主动调用adminClient.close()释放连接，避免资源泄漏。
  参考实现代码（Scala）：

import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig, OffsetSpec}
import org.apache.kafka.common.TopicPartition
import java.util.Properties

// Driver端全局初始化AdminClient
val adminProps = new Properties()
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-cluster:9092")
// 开启认证的集群在这里补充对应SSL/SASL配置即可
val adminClient = AdminClient.create(adminProps)

// 流处理逻辑中每个批次计算lag
val streamDF = spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "kafka-cluster:9092")
  .option("subscribe", "target_topic")
  .load()

streamDF.writeStream.foreachBatch((batchDF: DataFrame, batchId: Long) => {
  // 1. 从batchDF的Kafka元数据中解析当前批次处理完成后，每个分区的已消费offset
  val processedOffsets: Map[TopicPartition, Long] = // 自行解析得到的分区->已消费offset映射
  // 2. 调用Kafka原生接口拿实时最新offset
  val queryParam = processedOffsets.keys.map(tp => tp -> OffsetSpec.latest()).toMap.asJava
  val latestOffsetResult = adminClient.listOffsets(queryParam).all().get()
  // 3. 计算各分区lag
  val lagMetrics = processedOffsets.map { case (tp, consumedOffset) =>
    val realtimeLatestOffset = latestOffsetResult.get(tp).offset()
    (tp, realtimeLatestOffset - consumedOffset)
  }
  // 后续可将lagMetrics上报到监控系统、打日志等
}).start()

// 应用停止时释放资源
sys.addShutdownHook({
  adminClient.close()
})

常见踩坑提醒 #

• 不要直接用QueryProgress中自带的endOffset计算lag，这个值是批次拉数起点的偏移量，不是你处理完批次时的实时偏移量，结果会有明显偏差。
• 如果Kafka集群开了权限控制，记得给Spark任务使用的账号授予目标主题的DESCRIBE权限，否则调用Admin接口会报权限拒绝错误。
• 不要在executor端的算子逻辑里调用AdminClient拉取偏移量，会产生大量冗余的Kafka请求，给broker造成额外压力。

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