Spark Streaming集成Kafka问题：无法接收全部Kafka消息

我来帮你排查下这个问题，结合接收器模式的特性，整理几个常见的原因和对应的解决办法：

1. 端口配置错误（最容易忽略的点） #

你代码里的metadata.broker.list用了9042端口——这大概率是个低级错误！Kafka的默认监听端口是9092，而9042是Cassandra的默认端口。如果你的Kafka集群没修改过端口配置，这个错误会导致Spark无法正确连接到Kafka Broker，自然接收不全消息。赶紧把端口改成9092试试：

val kafkaConf = Map(
  "metadata.broker.list" -> "ip1:9092,ip2:9092",
  "group.id" -> "raw-email-event-streaming-consumer",
  "zookeeper.connect" -> "ip1:2181,ip2:2181"
)

2. 消费线程数与Kafka分区数不匹配 #

在接收器模式中，createStream第三个参数Map(RN_MAIL_TOPIC -> RN_MAIL...)里的数值是该主题的消费线程数。如果这个线程数小于Kafka主题的分区数，就会有部分分区的消息没人消费，直接导致接收不全。

解决办法：确保消费线程数至少等于Kafka主题的分区数（一般建议和分区数一致）。比如你的主题有3个分区，就设置为：

Map(RN_MAIL_TOPIC -> 3)

如果需要更高并行度，还可以创建多个接收器DStream，再用union合并，让每个接收器负责一部分分区。

3. 偏移量重置策略导致漏消费 #

如果你的Spark程序启动时间晚于Kafka消息的生产时间，而auto.offset.reset默认是latest，程序会从最新的偏移量开始消费，直接跳过之前已生产的消息，看起来就像“接收不全”。

你可以在kafkaConf里添加这个配置，设置为earliest，让程序从最早的未消费偏移量开始消费：

val kafkaConf = Map(
  // 其他配置...
  "auto.offset.reset" -> "earliest"
)

4. 未启用Write Ahead Logs（WAL）导致消息丢失 #

接收器模式下，如果没启用WAL，当Driver或Worker节点挂掉时，接收器已接收但还没处理的消息会直接丢失，重启后无法恢复。启用WAL可以把接收到的消息先写入分布式文件系统（比如HDFS），保证消息不丢失。

启用WAL的方法：

• 创建StreamingContext时开启WAL配置：

val conf = new SparkConf()
  .setAppName("KafkaStreaming")
  .set("spark.streaming.receiver.writeAheadLog.enable", "true")
val ssc = new StreamingContext(conf, Seconds(5))

• 同时在createStream时指定包含磁盘存储的级别：

val kafkaStream = KafkaUtils.createStream[Array[Byte], String, DefaultDecoder, StringDecoder](
  ssc, 
  kafkaConf, 
  Map(RN_MAIL_TOPIC -> 3), 
  StorageLevel.MEMORY_AND_DISK_SER
)

5. 推荐：迁移到Direct Stream模式（更可靠） #

其实Spark官方早就不推荐用接收器模式集成Kafka了，从Spark 1.3开始引入的Direct Stream（无接收器模式）更可靠高效：

• 直接从Kafka分区拉取消息，并行度和Kafka分区数完全匹配，不会出现接收器瓶颈
• 偏移量由Spark自己管理（可手动提交到Kafka或外部存储），避免ZooKeeper与Spark的偏移量不一致问题
• 不会出现消息堆积或丢失的情况

给你一个简单的Direct Stream示例：

import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "ip1:9092,ip2:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "raw-email-event-streaming-consumer",
  "auto.offset.reset" -> "earliest",
  "enable.auto.commit" -> (false: java.lang.Boolean)
)

val topics = Array(RN_MAIL_TOPIC)
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
)

// 处理消息+手动提交偏移量
stream.foreachRDD { rdd =>
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  // 这里写你的消息处理逻辑...
  // 处理完手动提交偏移量
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

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