Spark 2.4.X如何获取Kafka Headers？含Structured Streaming替代方案咨询

你说得没错，Spark 3.0版本才正式在spark-sql-kafka-0-10数据源中加入了对Kafka Headers的支持，而2.4.5版本确实没有原生支持——哪怕它依赖的kafka-client是2.0（已经兼容Kafka 0.11+的Headers特性），但Spark的数据源层并没有把这个字段暴露出来。下面给你两种可行的替代方案：

方案一：手动使用Kafka Consumer API读取并转换为DataFrame #

这种方法不需要修改Spark源码，直接利用Kafka的原生Consumer API读取消息，提取Headers后转成Spark DataFrame处理，适合快速实现需求。

实现步骤（Scala示例）： #

1. 初始化SparkSession和Kafka Consumer配置
2. 订阅目标Topic并拉取消息
3. 从ConsumerRecord中提取Headers，转成便于Spark处理的格式（比如Map）
4. 将消息数据封装为Tuple或Case Class，再转换为DataFrame

import org.apache.spark.sql.SparkSession
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, KafkaConsumer}
import scala.collection.JavaConverters._

// 初始化SparkSession
val spark = SparkSession.builder()
  .appName("KafkaHeadersReader")
  .master("local[*]") // 生产环境请去掉master配置
  .getOrCreate()

import spark.implicits._

// Kafka Consumer配置
val kafkaParams = Map(
  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "kafka-broker-1:9092,kafka-broker-2:9092",
  ConsumerConfig.GROUP_ID_CONFIG -> "spark-245-headers-consumer-group",
  ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
  ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
  ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
).asJava

// 创建Kafka Consumer并订阅Topic
val consumer = new KafkaConsumer[String, String](kafkaParams)
consumer.subscribe(List("your-target-topic").asJava)

// 拉取消息并处理Headers
val records = consumer.poll(3000).asScala.map { record =>
  // 将Kafka Headers转换为Map[String, String]（如果是二进制值可以保留BinaryType）
  val headersMap = record.headers().asScala
    .map(header => header.key() -> new String(header.value()))
    .toMap
  // 封装消息的关键信息
  (record.key(), record.value(), headersMap, record.offset(), record.timestamp())
}

// 转换为Spark DataFrame
val kafkaDF = records.toDF("key", "value", "headers", "offset", "timestamp")

// 查看结果
kafkaDF.show(false)

// 关闭资源
consumer.close()
spark.stop()

优缺点： #

• ✅ 优点：实现简单，无需修改Spark源码，快速验证需求
• ❌ 缺点：需要手动管理Kafka偏移量，如果要做流处理，得自己实现偏移量的持久化（比如存到数据库），无法直接利用Structured Streaming的Checkpoint机制

方案二：扩展Spark的Kafka数据源（自定义Source） #

如果需要在Structured Streaming中使用Headers，就需要扩展Spark原生的Kafka数据源，修改源码让它暴露Headers字段。

实现思路： #

1. 参考Spark 2.4.x的kafka010模块源码，复制KafkaSource、KafkaOffsetReader等核心类
2. 修改KafkaSource的getBatch方法，在构造返回的Row时，添加Headers字段（建议转成Array[Row]，每个Row包含key(String)和value(Binary)）
3. 更新数据源的Schema，增加headers字段，类型定义为：

   StructField("headers", ArrayType(StructType(Seq(
     StructField("key", StringType, nullable = false),
     StructField("value", BinaryType, nullable = true)
   ))), nullable = true)
   

4. 将修改后的代码打包成Jar，提交Spark作业时通过--jars参数加载，或者放到Spark的jars目录下
5. 使用自定义数据源读取Kafka流：

   val streamDF = spark.readStream
     .format("com.yourcompany.spark.sql.kafka010.CustomKafkaSource")
     .option("kafka.bootstrap.servers", "kafka-brokers:9092")
     .option("subscribe", "your-topic")
     .load()
   

优缺点： #

• ✅ 优点：可以完全集成到Structured Streaming中，自动管理偏移量和Checkpoint，适合生产环境的流处理场景
• ❌ 缺点：需要对Spark源码有一定了解，维护成本高，Spark版本升级时需要同步修改自定义Source的代码

补充：Spark Streaming（DStream）场景的处理 #

如果你的项目用的是传统的Spark Streaming（DStream），可以通过KafkaUtils.createDirectStream获取ConsumerRecord，然后提取Headers：

import org.apache.spark.streaming.kafka010._

val stream = KafkaUtils.createDirectStream[String, String](
  streamingContext,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](List("your-topic"), kafkaParams)
)

val streamWithHeaders = stream.map { record =>
  val headers = record.headers().asScala.map(h => h.key() -> new String(h.value())).toMap
  (record.key(), record.value(), headers)
}

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