韓晶晶 嚴律 黃春超
簡介
Spark Streaming是Spark Core的擴展，是構(gòu)建于Spark Core之上的實時流處理系統(tǒng)。相對于其他實時流處理系統(tǒng)，Spark Streaming最大的優(yōu)勢在于其位于Spark技術(shù)棧中，也即流處理引擎與數(shù)據(jù)處理引擎在同一個軟件棧中。在Spark Streaming中，數(shù)據(jù)的采集是以逐條方式，而數(shù)據(jù)處理是按批進行的。因此，其系統(tǒng)吞吐量會比流行的純實時流處理引擎Storm高2~5倍。
Spark Streaming對流數(shù)據(jù)處理的過成大致可以分為：啟動流處理引擎、接收和存儲流數(shù)據(jù)、處理流數(shù)據(jù)和輸出處理結(jié)果等四個步驟。其運行架構(gòu)圖如下所示：
[圖片上傳失敗...(image-f1cfaf-1542849231639)]
Step1 啟動流處理引擎
StreamingContext為Spark Streaming在Driver端的上下文，是spark streaming程序的入口。在該對象的啟 動過程中，會初始化其內(nèi)部的組件，其中最為重要的是DStreamGraph以及JobScheduler組件的初始化。
class StreamingContext private[streaming] (
_sc: SparkContext,
_cp: Checkpoint,
_batchDur: Duration
) extends Logging {
...
private[streaming] val conf = sc.conf
private[streaming] val env = sc.env
private[streaming] val graph: DStreamGraph = {
if (isCheckpointPresent) {
_cp.graph.setContext(this)
_cp.graph.restoreCheckpointData()
_cp.graph
} else {
require(_batchDur != null, "Batch duration for StreamingContext cannot be null")
val newGraph = new DStreamGraph()
newGraph.setBatchDuration(_batchDur)
newGraph
}
}
...
private[streaming] val scheduler = new JobScheduler(this)
...
}
Spark Streaming中作業(yè)的生成方式類似Spark核心，對DStream進行的各種操作讓他們之間構(gòu)建起依賴關(guān)系，DStreamGraph記錄了DStream之間的依賴關(guān)系等信息。
JobScheduler是Spark Streaming的Job總調(diào)度者。JobScheduler 有兩個非常重要的成員：JobGenerator 和 ReceiverTracker。JobGenerator維護一個定時器，定時為每個 batch 生成RDD DAG的實例；ReceiverTracker負責啟動、管理各個 receiver及管理各個receiver 接收到的數(shù)據(jù)。
通過調(diào)用StreamingContext#start()方法啟動流處理引擎。在StreamingContext#start()中，調(diào)用StreamingContext#validate()方法對DStreamGraph及checkpoint等做有效性檢查，然后啟動新的線程設置SparkContext，并啟動JobScheduler。
def start(): Unit = synchronized {
...
validate()
ThreadUtils.runInNewThread("streaming-start") {
sparkContext.setCallSite(startSite.get)
sparkContext.clearJobGroup()
sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false")
savedProperties.set(SerializationUtils.clone(sparkContext
.localProperties.get()))
scheduler.start()
}
state = StreamingContextState.ACTIVE
StreamingContext.setActiveContext(this)
...
}
Step2 接收與存儲流數(shù)據(jù)
JobScheduler啟動時，會創(chuàng)建一個新的 ReceiverTracker 實例 receiverTracker，并調(diào)用其start() 方法。在ReceiverTracker #start()中會初始化一個endpoint: ReceiverTrackerEndpoint對象，該對象用于接收和處理ReceiverTracker和 receivers之間 發(fā)送的消息。此外，在ReceiverTracker#start()中還會調(diào)用 launchReceivers 將各個receivers 分發(fā)到 executors 上。
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
}
}
ReceiverTracker#launchReceivers()會從DStreamGraph.inputStreams 中抽取出receivers，也即數(shù)據(jù)接收器。得到receivers后，給消息接收處理器 endpoint 發(fā)送 StartAllReceivers(receivers)消息。
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))
}
endpoint在接收到消息后，首先會判別消息的類型，對不同的消息執(zhí)行不同的處理操作。當收到StartAllReceivers類型的消息時，首先會計算每一個receiver要發(fā)送的目的executors，其計算主要遵循兩條原則：一是盡可能的使receiver分布均勻；二是如果receiver本身的preferredLocation不均勻，則以preferredLocation為準。然后遍歷每一個receiver，根據(jù)計算出的executors調(diào)用startReceiver方法來啟動receivers。
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)
}
由于ReceiverInputDStream實例只有一個receiver，但receiver可能需要在多個worker上啟動線程來接收數(shù)據(jù)，因此在startReceiver中需要將receiver及其對應的目的excutors轉(zhuǎn)換成RDD。
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))
}
轉(zhuǎn)換為RDD后，需要把receiver所進行的計算定義為startReceiverFunc函數(shù)，該函數(shù)以receiver實例為參數(shù)構(gòu)造ReceiverSupervisorImpl實例supervisor，構(gòu)造完畢后使用新線程啟動該supervisor并阻塞該線程。
val supervisor = new ReceiverSupervisorImpl(
receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
supervisor.start()
supervisor.awaitTermination()
最后，將receiverRDD以及要在receiverRDD上執(zhí)行的函數(shù)作為Job提交，以真正在各個executors上啟動Receiver。Job執(zhí)行后將會持續(xù)的進行數(shù)據(jù)的接收。
val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
Receiver源源不斷的接收到實時流數(shù)據(jù)后，根據(jù)接收數(shù)據(jù)的大小進行判斷，若數(shù)據(jù)量很小，則會聚集多條數(shù)據(jù)成一塊，然后進行塊存儲；若數(shù)據(jù)量很大，則直接進行塊存儲。對于這些數(shù)據(jù)，Receiver會直接交由ReceiverSupervisor，由其進行數(shù)據(jù)的轉(zhuǎn)儲操作。配置參數(shù)spark.streaming.receiver.writeAheadLog.enable的值決定是否預寫日志。根據(jù)參數(shù)值會產(chǎn)生不同類型的存儲receivedBlockHandler對象。
private val receivedBlockHandler: ReceivedBlockHandler = {
if (WriteAheadLogUtils.enableReceiverLog(env.conf)) {
//先寫 WAL，再存儲到 executor 的內(nèi)存或硬盤
new WriteAheadLogBasedBlockHandler(env.blockManager, receiver.streamId,
receiver.storageLevel, env.conf, hadoopConf, checkpointDirOption.get)
} else {
//直接存到 executor 的內(nèi)存或硬盤
new BlockManagerBasedBlockHandler(env.blockManager, receiver.storageLevel)
}
}
根據(jù)receivedBlockHandler進行塊存儲。將 block 存儲之后，會獲得 block 描述信息 blockInfo:ReceivedBlockInfo，這其中包含：streamId、數(shù)據(jù)位置、數(shù)據(jù)條數(shù)、數(shù)據(jù) size 等信息。接著，封裝以 block 作為參數(shù)的 AddBlock(blockInfo) 消息并發(fā)送給 ReceiverTracker 以通知其有新增 block 數(shù)據(jù)塊。
//調(diào)用 receivedBlockHandler.storeBlock 方法存儲 block，并得到一個 blockStoreResult
val blockStoreResult = receivedBlockHandler.storeBlock(blockId, receivedBlock)
//使用blockStoreResult初始化一個ReceivedBlockInfo實例
val blockInfo = ReceivedBlockInfo(streamId, numRecords, metadataOption, blockStoreResult)
//發(fā)送消息通知 ReceiverTracker 新增并存儲了 block
trackerEndpoint.askWithRetry[Boolean](AddBlock(blockInfo))
ReceiverTracker再把這些信息轉(zhuǎn)發(fā)給ReceivedBlockTracker，由其負責管理收到數(shù)據(jù)塊元信息。
private def addBlock(receivedBlockInfo: ReceivedBlockInfo): Boolean = {
receivedBlockTracker.addBlock(receivedBlockInfo)
}
step3 處理流數(shù)據(jù)
JobScheduler中有兩個主要的成員，一個是上文提到的ReceiverTracker，另一個則是JobGenerator 。在JobScheduler啟動時，會創(chuàng)建一個新的 JobGenerator 實例 jobGenerator，并調(diào)用其start() 方法。在 JobGenerator 的主構(gòu)造函數(shù)中，會創(chuàng)建一個定時器：
private val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds,
longTime => eventLoop.post(GenerateJobs(new Time(longTime))), "JobGenerator")
定時器中定義了批處理時間間隔ssc.graph.batchDuration.milliseconds。每當批處理時間到來時，會執(zhí)行一次eventLoop.post(GenerateJobs(new Time(longTime)))方法來向 eventLoop 發(fā)送 GenerateJobs(new Time(longTime))消息，eventLoop收到消息后會基于當前batch內(nèi)的數(shù)據(jù)進行Job的生成及提交執(zhí)行。
private def generateJobs(time: Time) {
// Checkpoint all RDDs marked for checkpointing to ensure their lineages are
// truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")
Try {
// allocate received blocks to batch
jobScheduler.receiverTracker.allocateBlocksToBatch(time)
// generate jobs using allocated block
graph.generateJobs(time)
} match {
case Success(jobs) =>
val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
case Failure(e) =>
jobScheduler.reportError("Error generating jobs for time " + time, e)
PythonDStream.stopStreamingContextIfPythonProcessIsDead(e)
}
eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
}
由源碼可知，eventLoop 在接收到 GenerateJobs(new Time(longTime))消息后首先調(diào)用了allocateBlocksToBatch()方法將已收到的blocks分配給batch。緊接著調(diào)用DStreamGraph類中的generateJobs()方法來生成基于該batch的Job序列。然后將批處理時間time、作業(yè)序列Seq[Job]和本批次數(shù)據(jù)的源信息包裝為JobSet，調(diào)用JobScheduler.submitJobSet(JobSet)提交給JobScheduler，JobScheduler將這些作業(yè)發(fā)送給Spark核心進行處理。
Step4 輸出處理結(jié)果
由于數(shù)據(jù)的處理有Spark核心來完成，因此處理的結(jié)果會從Spark核心中直接輸出至外部系統(tǒng)，如數(shù)據(jù)庫或者文件系統(tǒng)等，同時輸出的數(shù)據(jù)也可以直接被外部系統(tǒng)所使用。由于實時流數(shù)據(jù)的數(shù)據(jù)源源不斷的流入，Spark會周而復始的進行數(shù)據(jù)的計算，相應也會持續(xù)輸出處理結(jié)果。
