Spark定制班第9课:Spark Streaming源码解读之Receiver在Driver的精妙实现全生命周期彻底研究和思考

时间:2022-08-29 23:12:27
本期内容:
1. Receiver启动的方式设想
2. Receiver启动源码彻底分析

1. Receiver启动的方式设想

  Spark Streaming是个运行在Spark Core上的应用程序。这个应用程序既要接收数据,还要处理数据,这些都是在分布式的集群中进行的,应该启动多个Job,让它们分工并能协调。Receiver的工作是接收数据,应该是用Spark Core中的Job来实现。
  Receiver启动的设计,还要解决以下问题:
  1. 一个Executor上启动多个Receiver、而其它Executor却空闲的负载不均衡问题;
  2. Receiver启动异常导致整个Spark Streaming应用程序失败的问题。

2. Receiver启动源码彻底分析

  Spark Streaming的应用程序要处理流数据,肯定是在开始阶段就要做好接收数据的准备。
  Spark Streaming的应用程序代码定义DStream时,会定义一个或多个InputDStream。每个InputDStream分别对应有一个Receiver。

  Receiver启动全生命周期主流程图如下: 
Spark定制班第9课:Spark Streaming源码解读之Receiver在Driver的精妙实现全生命周期彻底研究和思考

  Receiver的启动,是在ssc.start()中。
  剖析一下StreamingContext的start():

/**
 * Start the execution of the streams.
 *
 * @throws IllegalStateException if the StreamingContext is already stopped.
 */
def start(): Unit = synchronized {
  state match {
    case INITIALIZED =>
      startSite.set(DStream.getCreationSite())
      StreamingContext.ACTIVATION_LOCK.synchronized {
        StreamingContext.assertNoOtherContextIsActive()
        try {
          validate()

          // Start the streaming scheduler in a new thread, so that thread local properties
          // like call sites and job groups can be reset without affecting those of the
          // current thread.
          ThreadUtils.runInNewThread("streaming-start") {
            sparkContext.setCallSite(startSite.get)
            sparkContext.clearJobGroup()
            sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false")
             // 启动子线程,一方面为了本地初始化工作,另外一方面是不要阻塞主线程。
             scheduler.start ()
          }
          state = StreamingContextState.ACTIVE
        } catch {
          case NonFatal(e) =>
            logError("Error starting the context, marking it as stopped", e)
            scheduler.stop(false)
            state = StreamingContextState.STOPPED
            throw e
        }
        StreamingContext.setActiveContext(this)
      }
      shutdownHookRef = ShutdownHookManager.addShutdownHook(
        StreamingContext.SHUTDOWN_HOOK_PRIORITY)(stopOnShutdown)
      // Registering Streaming Metrics at the start of the StreamingContextassert(env.metricsSystem != null)
      env.metricsSystem.registerSource(streamingSource)
      uiTab.foreach(_.attach())
      logInfo("StreamingContext started")
    case ACTIVE =>
      logWarning("StreamingContext has already been started")
    case STOPPED =>
      thrownew IllegalStateException("StreamingContext has already been stopped")
  }
}

  而在JobScheduler的start方法中ReceiverTracker的start方法被调用,Receiver就启动了。
  JobScheduler的start:

def start(): Unit = synchronized {
  if (eventLoop != null) return // scheduler has already been started

  logDebug("Starting JobScheduler")
  eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
    override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)

    override protected def onError(e: Throwable): Unit = reportError("Error in job scheduler", e)
  }
  eventLoop.start()

  // attach rate controllers of input streams to receive batch completion updates
  for {
    inputDStream <- ssc.graph.getInputStreams
    rateController <- inputDStream.rateController
  } ssc.addStreamingListener(rateController)

  listenerBus.start(ssc.sparkContext)
  receiverTracker =  new ReceiverTracker (ssc)
  inputInfoTracker = new InputInfoTracker(ssc)
   //启动receiverTracker
   receiverTracker.start ()
  jobGenerator.start()
  logInfo("Started JobScheduler")
}

  ReceiverTracker的start方法启动RPC消息通信体,为啥呢?因为ReceiverTracker会监控整个集群中的Receiver,Receiver转过来要向ReceiverTrackerEndpoint汇报自己的状态,接收的数据,包括生命周期等信息。
  ReceiverTracker.start:

/** Start the endpoint and receiver execution thread. */
def start(): Unit = synchronized {
  if (isTrackerStarted) {
    thrownew SparkException("ReceiverTracker already started")
  }
// Receiver的启动是依据输入数据流的。
if (!receiverInputStreams.isEmpty) {
     endpoint  = ssc.env.rpcEnv. setupEndpoint (
      "ReceiverTracker", new ReceiverTrackerEndpoint(ssc.env.rpcEnv))
    if (!skipReceiverLaunch)  launchReceivers ()
    logInfo("ReceiverTracker started")
    trackerState = Started
  }
}

  基于ReceiverInputDStream(是在Driver端)来获得具体的Receivers实例,然后再把他们分布到Worker节点上。一个ReceiverInputDStream只对应一个Receiver。
  ReceiverTracker.launchReceivers:

/**
 * Get the receivers from the ReceiverInputDStreams, distributes them to the
 * worker nodes as a parallel collection, and runs them.
 */
private def launchReceivers(): Unit = {
  val receivers = receiverInputStreams.map(nis => {
     // 一个数据输入来源(receiverInputDStream)只对应一个Receiver
     val  rcvr = nis.getReceiver()
    rcvr.setReceiverId(nis.id)
    rcvr
  })

   runDummySparkJob ()

  logInfo("Starting " + receivers.length + " receivers")
   // 此时的endpoint就是上面代码中在ReceiverTracker的start方法中构造的ReceiverTrackerEndpoint
   endpoint.send(StartAllReceivers(receivers))
}

  先看其中的runDummySparkJob()。
  runDummySparkJob()是为了确保所有节点活着,而且避免所有的receivers集中在一个节点上。
  ReceiverTracker.runDummySparkJob():

/**
 * Run the dummy Spark job to ensure that all slaves have registered. This avoids all the
 * receivers to be scheduled on the same node.
 *
 * TODO Should poll the executor number and wait for executors according to
 * "spark.scheduler.minRegisteredResourcesRatio"and
 * "spark.scheduler.maxRegisteredResourcesWaitingTime" rather than running a dummy job.
 */
private def runDummySparkJob(): Unit = {
  if (!ssc.sparkContext.isLocal) {
    ssc.sparkContext.makeRDD(1 to 50, 50).map(x => (x, 1)).reduceByKey(_ + _, 20).collect()
  }
  assert(getExecutors.nonEmpty)
}

  再回去看ReceiverTracker.launchReceivers()中的getReceiver()。

  ReceiverInputDStream.getReceiver():

/**
 * Gets the receiver object that will be sent to the worker nodes
 * to receive data. This method needs to defined by any specific implementation
 * of a ReceiverInputDStream.
 */
def getReceiver(): Receiver[T]  //返回的是Receiver对象

  ReceiverInputDStream的getReceiver()方法返回Receiver对象。 该方法实际上要靠ReceiverInputDStream的子类实现。
  相应的,ReceiverInputDStream的子类中必须要实现这个getReceiver()方法。ReceiverInputDStream的子类还必须定义自己对应的Receiver子类,因为这个Receiver子类会在getReceiver()方法中用来创建这个Receiver子类的对象。
  根据继承关系,这里看一下ReceiverInputDStream的子类SocketInputDStream中的getReceiver方法。
  SocketInputDStream.getReceiver:

def getReceiver(): Receiver[T] = {
    new  SocketReceiver (host, port, bytesToObjects, storageLevel)
  }
}

  SocketInputDStream中还定义了相应的Receiver子类SocketReceiver。SocketReceiver类中还必须定义onStart方法。
  onStart方法会启动后台线程,调用receive方法。

private[streaming]
class SocketReceiver[T: ClassTag](
    host: String,
    port: Int,
    bytesToObjects: InputStream => Iterator[T],
    storageLevel: StorageLevel
  ) extends Receiver[T](storageLevel) with Logging {

  再回到 ReceiverTracker. launchReceivers() 中,看最后的代码  endpoint.send(StartAllReceivers(receivers))。这个代码给ReceiverTrackerEndpoint对象发送了StartAllReceivers消息,ReceiverTrackerEndpoint对象接收后所做的处理在ReceiverTrackerEndpoint.receive中。

  ReceiverTracker.ReceiverTrackerEndpoint.receive:

/** RpcEndpoint to receive messages from the receivers. */
private class ReceiverTrackerEndpoint(override val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint {

  // TODO Remove this thread pool after https://github.com/apache/spark/issues/7385 is merged
private val submitJobThreadPool = ExecutionContext.fromExecutorService(
    ThreadUtils.newDaemonCachedThreadPool("submit-job-thread-pool"))

  privateval walBatchingThreadPool = ExecutionContext.fromExecutorService(
    ThreadUtils.newDaemonCachedThreadPool("wal-batching-thread-pool"))

  @volatile private var active: Boolean = true

  override def receive: PartialFunction[Any, Unit] = {
    // Local messages
    case  StartAllReceivers (receivers) =>
      // schedulingPolicy调度策略
      // receivers就是要启动的receiver
      // getExecutors获得集群中的Executors的列表
      // scheduleReceivers就可以确定receiver可以运行在哪些Executor上
      val scheduledLocations =  schedulingPolicy. scheduleReceivers (receivers, getExecutors)
      for (receiver <- receivers) {
      // scheduledLocations根据receiver的Id就找到了当前那些Executors可以运行
      Receiverval executors = scheduledLocations(receiver.streamId)
        updateReceiverScheduledExecutors(receiver.streamId, executors)
        receiverPreferredLocations(receiver.streamId) = receiver.preferredLocation
      // 上述代码之后要启动的Receiver确定了,具体Receiver运行在哪些Executors上也确定了。
      // 循环receivers,每次将一个receiver传入过去。
         startReceiver (receiver, executors)
      }
      // 用于接收RestartReceiver消息,重新启动Receiver.
    case RestartReceiver(receiver) =>
      // Old scheduled executors minus the ones that are not active any more
      // 如果Receiver失败的话,从可选列表中减去。
      // 刚在调度为Receiver分配给哪个Executor的时候会有一些列可选的Executor列表
      val oldScheduledExecutors = getStoredScheduledExecutors(receiver.streamId)
      // 重新获取Executors
      val scheduledLocations = if (oldScheduledExecutors.nonEmpty) {
          // Try global scheduling again
          oldScheduledExecutors
        } else {
           // 如果可选的Executor使用完了,则会重新执行rescheduleReceiver重新获取Executor.
          val oldReceiverInfo = receiverTrackingInfos(receiver.streamId)
          // Clear "scheduledLocations" to indicate we are going to do local scheduling
          val newReceiverInfo = oldReceiverInfo.copy(
            state = ReceiverState.INACTIVE, scheduledLocations = None)
          receiverTrackingInfos(receiver.streamId) = newReceiverInfo
          schedulingPolicy.rescheduleReceiver(
            receiver.streamId,
            receiver.preferredLocation,
            receiverTrackingInfos,
            getExecutors)
        }
      // Assume there is one receiver restarting at one time, so we don't need to update
      // receiverTrackingInfos
       // 重复调用startReceiver
       startReceiver (receiver, scheduledLocations)
    case c: CleanupOldBlocks =>
      receiverTrackingInfos.values.flatMap(_.endpoint).foreach(_.send(c))
    case UpdateReceiverRateLimit(streamUID, newRate) =>
      for (info <- receiverTrackingInfos.get(streamUID); eP <- info.endpoint) {
        eP.send(UpdateRateLimit(newRate))
      }
    // Remote messagescase ReportError(streamId, message, error) =>
      reportError(streamId, message, error)
  }

  从注释中可以看到,Spark Streaming指定receiver在哪些Executors上运行,而不是基于Spark Core中的Task来指定。
  Spark使用submitJob的方式启动Receiver,而在应用程序执行的时候会有很多Receiver,这个时候是启动一个Receiver呢,还是把所有的Receiver通过这一个Job启动? 
  在ReceiverTracker的receive方法中startReceiver方法第一个参数就是receiver,从实现中可以看出for循环不断取出receiver,然后调用startReceiver。由此就可以得出一个Job只启动一个Receiver。
  如果Receiver启动失败,此时并不会认为是作业失败,会重新发消息给ReceiverTrackerEndpoint重新启动Receiver,这样也就确保了Receivers一定会被启动,这样就不会像Task启动Receiver的话如果失败受重试次数的影响。
  ReceiverTracker.startReceiver:

    /**
     * Start a receiver along with its scheduled executors
     */
    private def startReceiver(
        receiver: Receiver[_],
        // scheduledLocations指定的是在具体的那台物理机器上执行。
        scheduledLocations: Seq[TaskLocation]): Unit = {
      // 判断下Receiver的状态是否正常。
      def shouldStartReceiver: Boolean = {
        // It's okay to start when trackerState is Initialized or Started
        !(isTrackerStopping || isTrackerStopped)
      }

      val receiverId = receiver.streamId
      if (!shouldStartReceiver) {
        // 如果不需要启动Receiver则会调用
        onReceiverJobFinish(receiverId)
        return
      }

      val checkpointDirOption = Option(ssc.checkpointDir)
      val serializableHadoopConf =
        new SerializableConfiguration(ssc.sparkContext.hadoopConfiguration)

      // startReceiverFunc封装了在worker上启动receiver的动作。
      // Function to start the receiver on the worker node
      val  startReceiverFunc : Iterator[Receiver[_]] => Unit =
        (iterator: Iterator[Receiver[_]]) => {
          if (!iterator.hasNext) {
            throw new SparkException(
              "Could not start receiver as object not found.")
          }
          if (TaskContext.get().attemptNumber() == 0) {
            val receiver = iterator.next()
            assert(iterator.hasNext == false)
            // ReceiverSupervisorImpl是Receiver的监控器,同时负责数据的写等操作。
            val supervisor = new ReceiverSupervisorImpl(
              receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
             supervisor.start ()
            supervisor.awaitTermination()
          } else {
              // 如果你想重新启动receiver的话,你需要重新完成上面的调度,重新schedule,而不是Task重试。
            // It's restarted by TaskScheduler, but we want to reschedule it again. So exit it.
          }
        }

      // Create the RDD using the scheduledLocations to run the receiver in a Spark job
      val receiverRDD: RDD[Receiver[_]] =
        if (scheduledLocations.isEmpty) {
          ssc.sc.makeRDD(Seq(receiver), 1)
        } else {
          val preferredLocations = scheduledLocations.map(_.toString).distinct
          ssc.sc.makeRDD(Seq(receiver -> preferredLocations))
        }
      // receiverId可以看出,receiver只有一个
      receiverRDD.setName(s"Receiver $receiverId")
      ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
      ssc.sparkContext.setCallSite(Option(ssc.getStartSite()).getOrElse(Utils.getCallSite()))

      // 每个Receiver的启动都会触发一个Job,而不是一个作业的Task去启动所有的Receiver.
      // 应用程序一般会有很多Receiver,
      // 调用SparkContext的submitJob,为了启动Receiver,启动了Spark一个作业。
      val future =  ssc.sparkContext.submitJob [Receiver[_], Unit, Unit](
        receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
      // We will keep restarting the receiver job until ReceiverTracker is stopped
      future.onComplete {
        case Success(_) =>
          if (!shouldStartReceiver) {
            onReceiverJobFinish(receiverId)
          } else {
            logInfo(s"Restarting Receiver $receiverId")
            self.send(RestartReceiver(receiver))
          }
        case Failure(e) =>
          if (!shouldStartReceiver) {
            onReceiverJobFinish(receiverId)
          } else {
            logError("Receiver has been stopped. Try to restart it.", e)
            logInfo(s"Restarting Receiver $receiverId")
            self.send( RestartReceiver (receiver))
          }
      // 使用线程池的方式提交Job,这样的好处是可以并发的启动Receiver。
      }(submitJobThreadPool)
      logInfo(s"Receiver ${receiver.streamId} started")
    }


  当Receiver启动失败的话,就会触发ReceiverTrackEndpoint重新启动一个Spark Job去启动Receiver.

/**
 * This message will trigger ReceiverTrackerEndpoint to restart a Spark job for the receiver.
 */
private[streaming] case class RestartReceiver(receiver: Receiver[_]) extends ReceiverTrackerLocalMessage
// 当Receiver关闭的话,并不需要重新启动Spark Job.
/**
 * Call when a receiver is terminated. It means we won't restart its Spark job.
 */
private def onReceiverJobFinish(receiverId: Int): Unit = {
  receiverJobExitLatch.countDown()
// 使用foreach将receiver从receiverTrackingInfo中去掉。
  receiverTrackingInfos.remove(receiverId).foreach { receiverTrackingInfo =>
    if (receiverTrackingInfo.state == ReceiverState.ACTIVE) {
      logWarning(s"Receiver $receiverId exited but didn't deregister")
    }
  }
}

  回头再看ReceiverTracker.startReceiver中的代码supervisor.start()。在子类ReceiverSupervisorImpl中并没有start方法,因此调用的是父类ReceiverSupervisor的start方法。
  ReceiverSupervisor.start:

/** Start the supervisor */
def start() {
   onStart () // 具体实现是子类实现的。
   startReceiver ()
}

  Receiver的onStart方法源码如下:

/**
 * Called when supervisor is started.
 * Note that this must be called before the receiver.onStart() is called to ensure
 * things like [[BlockGenerator]]s are started before the receiver starts sending data.
 */
protected def onStart() { }

  其具体实现是在子类的ReceiverSupervisorImpl的onStart方法:

overrideprotected def onStart() {
  registeredBlockGenerators.foreach { _. start () }
}

  其中的_.start()是BlockGenerator.start:

/** Start block generating and pushing threads. */
def start(): Unit = synchronized {
  if (state == Initialized) {
    state = Active
    blockIntervalTimer.start()
    blockPushingThread.start()
    logInfo("Started BlockGenerator")
  } else {
    thrownew SparkException(
      s"Cannot start BlockGenerator as its not in the Initialized state [state = $state]")
  }
}

  回过头再看ReceiverSupervisor.start中的startReceiver()。
  ReceiverSupervisor.startReceiver:

  /** Start receiver */
  def startReceiver(): Unit = synchronized {
    try {
      if (onReceiverStart()) {
        logInfo("Starting receiver")
        receiverState = Started
         receiver.onStart ()
        logInfo("Called receiver onStart")
      } else {
        // The driver refused us
        stop("Registered unsuccessfully because Driver refused to start receiver " + streamId, None)
      }
    } catch {
      case NonFatal(t) =>
        stop("Error starting receiver " + streamId, Some(t))
    }
  }

  仍以Receiver的子类SocketReceiver为例说明onStart方法:
  SocketReceiver.onStart:

  def  onStart () {
    // Start the thread that receives data over a connection
    new Thread("Socket Receiver") {
      setDaemon(true)
      override def run() {  receive () }
    }.start()
  }

  这个onStart方法开启了的线程,用于启动socket来接收数据。这个被运行的receive()被定义在 ReceiverInputDStream的 子类SocketInputDStream中。
  SocketInputDStream.receive:

/** Create a socket connection and receive data until receiver is stopped */
def receive() {
    var socket: Socket = null
    try {
      logInfo("Connecting to " + host + ":" + port)
      socket =  new Socket (host, port)
      logInfo("Connected to " + host + ":" + port)
      val iterator = bytesToObjects(socket.getInputStream())
      while(!isStopped && iterator.hasNext) {
        store(iterator.next)
      }
      if (!isStopped()) {
        restart("Socket data stream had no more data")
      } else {
        logInfo("Stopped receiving")
      }
    } catch {
      case e: java.net.ConnectException =>
        restart("Error connecting to " + host + ":" + port, e)
      case NonFatal(e) =>
        logWarning("Error receiving data", e)
        restart("Error receiving data", e)
    } finally {
      if (socket != null) {
        socket.close()
        logInfo("Closed socket to " + host + ":" + port)
      }
    }
  }
}