Spark Streaming读取AMQ topic无数据时如何自动停止流任务

问题根源 #

你在Worker端修改的ActiveMQReceiver.stop是Worker进程内的静态变量，和Driver进程完全隔离。Driver端的CustomReceiver实例是初始化时生成的本地副本，永远无法感知到Worker端的变量变更，所以调用checkStatus永远返回false。
普通累加器不生效通常是因为你没有在Driver端初始化累加器、再将实例传入接收器，而是在接收器内部直接初始化，此时累加器是Worker端的本地实例，更新不会同步到Driver。

可行解决方案 #

方案1：哨兵消息法（实现最简单，无延迟） #

通过在无数据时往流中写入特殊标记消息，Driver端直接从流数据中感知停止信号，不需要跨进程共享变量。

1. 修改接收器的无数据分支逻辑，写入哨兵消息：

private def receive() {
  activeMQStream = new ActiveMQStream(broker, topic, ...)
  val topicSubscriber = activeMQStream.getTopicSubscriber()
  // 定义哨兵消息，确保和正常业务数据不重复
  val SENTINEL_STOP = "__AMQ_NO_MORE_DATA__"

  while(!isStopped && !ActiveMQReceiver.stop){
     val message = topicSubscriber.receive(timeOutInMilliseconds)
     if (message != null && message.isInstanceOf[TextMessage]) {
         val textMessage = message.asInstanceOf[TextMessage];
         val text = textMessage.getText();
         store(text)
         println("ActiveMQReceiver: there is data from AMQ ....")
     } else {
         // 先写入哨兵消息通知Driver
         store(SENTINEL_STOP)
         ActiveMQReceiver.stop = true
         println("ActiveMQReceiver: No more data from AMQ .....")
     }
}

2. 修改Driver端的foreachRDD逻辑，识别哨兵消息设置停止标记：

// Driver端提前定义和接收器一致的哨兵
val SENTINEL_STOP = "__AMQ_NO_MORE_DATA__"
@volatile var stopFlag = false

stream.foreachRDD { rdd =>
  if(rdd.count() > 0){
    val fromWorker = rdd.collect().toList
    // 检查是否包含停止哨兵
    if (fromWorker.contains(SENTINEL_STOP)) {
      // 过滤哨兵后合并正常数据
      driverList = driverList ::: fromWorker.filter(_ != SENTINEL_STOP)
      // 直接设置Driver端的停止标记
      stopFlag = true
    } else {
      driverList = driverList ::: fromWorker
    }
  }
} 

3. 原有轮询逻辑直接判断Driver本地的stopFlag即可，不需要再调用customReceiver.checkStatus()。

方案2：累加器法（适合不想修改数据链路的场景） #

用Spark原生的累加器实现Worker到Driver的单向状态同步，注意必须在Driver端初始化累加器再传入接收器：

1. Driver端初始化累加器，传入自定义接收器构造参数：

// 初始化累加器，只能Driver读，Worker写
val noDataAccum = spark.sparkContext.longAccumulator("amq_no_data_flag")
val customReceiver = new CustomReceiver(brokerURL, topic, noDataAccum, ...)

2. 接收器接收累加器参数，无数据时更新累加器：

class CustomReceiver(brokerURL: String, topic: String, noDataAccum: LongAccumulator, ...){
    // ...原有逻辑不变
    private def receive() {
      activeMQStream = new ActiveMQStream(broker, topic, ...)
      val topicSubscriber = activeMQStream.getTopicSubscriber()

      while(!isStopped && !ActiveMQReceiver.stop){
         val message = topicSubscriber.receive(timeOutInMilliseconds)
         if (message != null && message.isInstanceOf[TextMessage]) {
             val textMessage = message.asInstanceOf[TextMessage];
             val text = textMessage.getText();
             store(text)
             println("ActiveMQReceiver: there is data from AMQ ....")
         } else {
             // 更新累加器通知Driver
             noDataAccum.add(1)
             ActiveMQReceiver.stop = true
             println("ActiveMQReceiver: No more data from AMQ .....")
         }
    }
}

3. Driver端轮询时读取累加器值判断停止条件：

while (!isStopped) {
  isStopped = ssc.awaitTerminationOrTimeout(checkIntervalMillis)
  println("Check if stop flag was raised")
  // 读取累加器值判断是否停止
  stopFlag = noDataAccum.value > 0

  if (!isStopped && stopFlag) {
    var seq = driverList.toSeq
    import spark.implicits._
    val df = seq.toDS()
    println("Request to stop")
    ssc.stop(false, true)
  }
}

注意事项 #

• AMQ的receive超时时间要设置合理，避免正常消息间隔被误判为无数据，也避免过长导致停止延迟。
• 如果需要避免网络波动导致的误判，可以设置连续N个批次无数据再触发停止逻辑。

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