——————————————————————————————
Spark Streaming概述
Spark Streaming 初始化过程
Spark Streaming Receiver启动过程分析
Spark Streaming 数据准备阶段分析（Receiver方式）
Spark Streaming 数据计算阶段分析
SparkStreaming Backpressure分析
Spark Streaming Executor DynamicAllocation 机制分析
——————————————————————————————
Receiver是数据准备阶段的一个主要组件，其负载接入外部数据，其生命周期由ReceiverTracker负责管理。
Receiver的启动
1. Receiver抽取与Executor准备
“Spark Streaming 初始化过程”中提到 JobScheduler在启动时会创建和启动ReceiverTracker.
在ReceiverTracker创建时，其会从DStreamGraph中抽取出ReceiverInputStream，以便在启动Receiver时从中抽取出Receiver,然后一一启动。
private val receiverInputStreams = ssc.graph.getReceiverInputStreams()
在ReceiverTracker启动时，其主要做如下两件事：
- 创建ReceiverTrackerEndpoint，用于接收Receiver的信息
- 启动Receiver.
ReceiverTracker的Start方法如下所示：
/** Start the endpoint and receiver execution thread. */
def start(): Unit = synchronized {
if (isTrackerStarted) {
throw new SparkException("ReceiverTracker already started")
}
if (!receiverInputStreams.isEmpty) {
endpoint = ssc.env.rpcEnv.setupEndpoint(
"ReceiverTracker", new ReceiverTrackerEndpoint(ssc.env.rpcEnv))
if (!skipReceiverLaunch) launchReceivers()
logInfo("ReceiverTracker started")
trackerState = Started
}
}
其中 launchReceivers（）方法用于启动Receiver, 其代码如下：
/**
* 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 =>
val rcvr = nis.getReceiver()
rcvr.setReceiverId(nis.id)
rcvr
}
runDummySparkJob()
logInfo("Starting " + receivers.length + " receivers")
// 发送启动指令
endpoint.send(StartAllReceivers(receivers))
}
此方法的主要操作有：
- 从ReceiverInputStreams中抽取Receiver, 并将streamId做为Receiver的id.
- 执行runDummySparkJob，此方法是执行一个简单的SparkJob，目的是为确保应用申请的Executor的最小份额得以满足，最小份额由参数“spark.cores.max” 和 “spark.scheduler.minRegisteredResourcesRatio” 共同决定，默认为申请的所有Executor。当应用已获得的Executor数量小于最小份额时，Job将阻塞并等待Executor注册，直到满足其运行需要的最小限额。
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)
}
程序逻辑非常简单，目的是使其在不消耗过多资源的情况下，可以保证在调度Recevier时，已有大量的Executor注册完成，从而使Recevier调度时尽量均匀的调度至不同的Executor 。
- 向ReceiverTrackerEndpoint发送启动所有executor指令（StartAllReceivers）
在ReceiverTrackerEndpoint收到StartAllReceivers指令后，其将
- 调度Receiver： 为Receiver设置执行位置信息
- 启动Receiver
其实现逻辑如下：
case StartAllReceivers(receivers) =>
val scheduledLocations = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
for (receiver <- receivers) {
val executors = scheduledLocations(receiver.streamId)
updateReceiverScheduledExecutors(receiver.streamId, executors)
receiverPreferredLocations(receiver.streamId) = receiver.preferredLocation
startReceiver(receiver, executors)
}
2. Receiver 调度
Receiver调度工作由ReceiverSchedulingPolicy进行，对Receiver的调度工作主要可以分为如下两个阶段：
- 全局调度阶段
此阶段发生在首次调度Receiver时，此阶段会保证receivers尽量均匀的分散在Executors中。调度过程中会为每一个Receiver指定启动的位置信息（location）
- 局部调度阶段
此阶段发生在Receiver重启时，仅需启动失败Receiver
全局调度阶段是必然会发生的，因此将以这种情况为例对Receiver调度进行详细说明。其调度过程如下：
- 获取所有executor的主要地址信息
- 创建numReceiversOnExecutor用于记录每个Executor分配的Receiver数目
- 创建scheduledLocations用于记录用户指定偏好位置的Receiver
- 调度指定preferredLocation信息的Receiver. 遍历Receivers, 为用户指定的preferredLocation的主机中选择启动Receiver数 最少的Executor做为当前Receiver启动位置，并更新记录scheduledLocations 和numReceiversOnExecutor。
- 调度未指定preferredLocation信息的Receiver.
将Executor依照分配的Receiver数目从小到大排序，为Receiver分配一个Executor.
- 若还有剩余Executor, 将这些Executor 加入到拥有最少候选对象的Receiver列表中。
至此， Receiver与与Executor的关联联系建立完毕。
调度的实现代码如下所示：
/**
* Try our best to schedule receivers with evenly distributed. However, if the
* `preferredLocation`s of receivers are not even, we may not be able to schedule them evenly
* because we have to respect them.
*
* Here is the approach to schedule executors:
* <ol>
* <li>First, schedule all the receivers with preferred locations (hosts), evenly among the
* executors running on those host.</li>
* <li>Then, schedule all other receivers evenly among all the executors such that overall
* distribution over all the receivers is even.</li>
* </ol>
*
* This method is called when we start to launch receivers at the first time.
*
* @return a map for receivers and their scheduled locations
*/
def scheduleReceivers(
receivers: Seq[Receiver[_]],
executors: Seq[ExecutorCacheTaskLocation]): Map[Int, Seq[TaskLocation]] = {
if (receivers.isEmpty) {
return Map.empty
}
if (executors.isEmpty) {
return receivers.map(_.streamId -> Seq.empty).toMap
}
val hostToExecutors = executors.groupBy(_.host)
val scheduledLocations = Array.fill(receivers.length)(new mutable.ArrayBuffer[TaskLocation])
val numReceiversOnExecutor = mutable.HashMap[ExecutorCacheTaskLocation, Int]()
// Set the initial value to 0
executors.foreach(e => numReceiversOnExecutor(e) = 0)
// Firstly, we need to respect "preferredLocation". So if a receiver has "preferredLocation",
// we need to make sure the "preferredLocation" is in the candidate scheduled executor list.
for (i <- 0 until receivers.length) {
// Note: preferredLocation is host but executors are host_executorId
receivers(i).preferredLocation.foreach { host =>
hostToExecutors.get(host) match {
case Some(executorsOnHost) =>
// preferredLocation is a known host. Select an executor that has the least receivers in
// this host
val leastScheduledExecutor =
executorsOnHost.minBy(executor => numReceiversOnExecutor(executor))
scheduledLocations(i) += leastScheduledExecutor
numReceiversOnExecutor(leastScheduledExecutor) =
numReceiversOnExecutor(leastScheduledExecutor) + 1
case None =>
// preferredLocation is an unknown host.
// Note: There are two cases:
// 1. This executor is not up. But it may be up later.
// 2. This executor is dead, or it's not a host in the cluster.
// Currently, simply add host to the scheduled executors.
// Note: host could be `HDFSCacheTaskLocation`, so use `TaskLocation.apply` to handle
// this case
scheduledLocations(i) += TaskLocation(host)
}
}
}
// For those receivers that don't have preferredLocation, make sure we assign at least one
// executor to them.
for (scheduledLocationsForOneReceiver <- scheduledLocations.filter(_.isEmpty)) {
// Select the executor that has the least receivers
val (leastScheduledExecutor, numReceivers) = numReceiversOnExecutor.minBy(_._2)
scheduledLocationsForOneReceiver += leastScheduledExecutor
numReceiversOnExecutor(leastScheduledExecutor) = numReceivers + 1
}
// Assign idle executors to receivers that have less executors
val idleExecutors = numReceiversOnExecutor.filter(_._2 == 0).map(_._1)
for (executor <- idleExecutors) {
// Assign an idle executor to the receiver that has least candidate executors.
val leastScheduledExecutors = scheduledLocations.minBy(_.size)
leastScheduledExecutors += executor
}
receivers.map(_.streamId).zip(scheduledLocations).toMap
}
此实现，存在一个问题，如果Receiver设置了preferredLocation且preferredLocation所对应的主机存在此应用的Executor的情况下，也不一定保证Receiver调度至此Executor.
3. Receiver 启动
在为Receiver设置完启动位置之后，将调用startReceiver方法启动Receiver, 启动过程如下：
- 依据preferredLocation将Receiver包装成RDD
// 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))
}
receiverRDD.setName(s"Receiver $receiverId")
- 以SparkJob的形式提交作业, Receiver作为Task 以线程方式执行
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))
}
}(ThreadUtils.sameThread)
- Task执行， 执行的startReceiverFunc方法，该方法会创建并启动ReceiverSupervisorImpl（Job及Task调度过程此处不再详细说明，同批处理）
// 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)
val supervisor = new ReceiverSupervisorImpl(
receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
supervisor.start()
supervisor.awaitTermination()
} else {
// It's restarted by TaskScheduler, but we want to reschedule it again. So exit it.
}
}
其中ReceiverSupervisorImpl 提供了处理Receiver接收数据的所有必要的方法。并且它还创建了BlockGenerator，用于对Receiver接收的数据流进行切片操作。
其ReceiverSupervisorImpl的Start方法实现如下：
/** Start the supervisor */
def start() {
onStart()
startReceiver()
}
其中onStart() 会创建BlockGenerator并启动。
startReceiver()方法，首先会向ReceiverTracker注册Receiver信息，并验证Receiver是否合法。若合法，则调用Receiver的onStart方法进行数据接收，其实现逻辑如下：
/** Start receiver */
def startReceiver(): Unit = synchronized {
try {
if (onReceiverStart()) {
logInfo(s"Starting receiver $streamId")
receiverState = Started
receiver.onStart()
logInfo(s"Called receiver $streamId 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))
}
}
下面以WordCount中的SocketInputDStream中的SocketReceiver为例进行说明,其onStart方法实现如下：
def onStart() {
logInfo(s"Connecting to $host:$port")
try {
socket = new Socket(host, port)
} catch {
case e: ConnectException =>
restart(s"Error connecting to $host:$port", e)
return
}
logInfo(s"Connected to $host:$port")
// Start the thread that receives data over a connection
new Thread("Socket Receiver") {
setDaemon(true)
override def run() { receive() }
}.start()
}
/** Create a socket connection and receive data until receiver is stopped */
def receive() {
try {
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 NonFatal(e) =>
logWarning("Error receiving data", e)
restart("Error receiving data", e)
} finally {
onStop()
}
}
}
通过上述实现可知，其将通过socket方式进行数据接收。
Receiver启动流程至此结束，Receiver启动之后会接收源源不断的数据流并对数据分片，副本分发工作，为计算阶段做准备，接下来将进行数据准备环节的分析。