Spark Streaming写入Kafka优化后程序卡顿无消息输出求助

针对你遇到的问题，结合Spark 1.6.0的特性和KafkaProducer的运行机制，我整理了几个关键排查方向和解决方案：

1. 核心问题：KafkaProducer无法通过广播变量正确序列化/反序列化 #

Spark 1.6.0的广播变量会将对象序列化后分发到Executor，但KafkaProducer本身包含不可序列化的网络连接、内部状态等组件。如果你直接广播已经初始化完成的KafkaProducer实例，Executor端拿到的会是一个失效的对象——序列化过程会断开网络连接，破坏Producer的内部状态，导致后续无法正常发送消息，最终触发Producer的超时关闭日志。

解决思路：不要广播Producer实例，改为广播Producer的配置参数，在Executor端的foreachPartition中延迟初始化Producer，确保每个Executor使用本地创建的可用实例。

2. 检查KafkaSink实现的线程安全与生命周期问题 #

虽然KafkaProducer本身是线程安全的，但如果你的Sink实现存在以下问题，也会导致卡顿或无消息：

• 错误地在Task执行结束后调用producer.close()：如果多个Task共享同一个广播的Producer实例，第一个Task结束后就会关闭Producer，后续Task无法再使用。
• 未处理Executor级别的Producer复用：广播方案如果没有做Executor级别的缓存，反而会比原foreachPartition方案更低效。

优化方案：改用Executor级别缓存Producer的方式，通过静态变量+双重检查锁实现每个Executor只创建一个Producer实例，既复用连接提升性能，又避免序列化问题：

object KafkaProducerHolder {
  @volatile private var producer: KafkaProducer[String, String] = null

  def getProducer(config: Properties): KafkaProducer[String, String] = {
    if (producer == null) {
      synchronized {
        if (producer == null) {
          producer = new KafkaProducer[String, String](config)
        }
      }
    }
    producer
  }
}

// 在Streaming任务中使用
stream.foreachPartition { partition =>
  val kafkaConfig = new Properties()
  kafkaConfig.put("bootstrap.servers", "your-broker-list")
  kafkaConfig.put("acks", "1")
  // 其他Kafka配置...
  
  val producer = KafkaProducerHolder.getProducer(kafkaConfig)
  partition.foreach { msg =>
    val record = new ProducerRecord[String, String]("target-topic", msg._1, msg._2)
    producer.send(record)
  }
  // 不要在这里关闭Producer，留给同一Executor的其他Task复用
}

3. 深挖Yarn日志的隐藏异常 #

你看到的Closing the Kafka producer with timeoutMillis...只是最终的关闭日志，建议过滤Executor级别的日志，查找以下关键异常：

• NotSerializableException：确认是否存在未序列化的对象被广播
• TimeoutException或连接失败日志：检查Producer初始化时的网络连通性、Broker地址是否正确
• IllegalStateException：查看是否存在Producer被重复关闭或已关闭后再次使用的情况

4. 对比原方案的优化空间 #

原foreachPartition创建Producer的方案本身是可行的，300K消息8秒的性能可以通过Executor级别的Producer复用进一步优化（即上面的KafkaProducerHolder方案），既保留了原方案的稳定性，又减少了Producer创建销毁的开销。

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