<一>DStream实时流数据处理
Spark Streaming能够对流数据进行近乎实时的速度进行数据处理。采用了不同于一般的流式数据处理模型，该模型使得Spark Streaming有非常高的处理速度，与storm相比拥有更高的吞能力。
本篇简要分析Spark Streaming的处理模型，Spark Streaming系统的初始化过程，以及当接收到外部数据时后续的处理步骤。
系统概述
流数据的特点
与一般的文件（即内容已经固定）型数据源相比，所谓的流数据拥有如下的特点
- 数据一直处在变化中
- 数据无法回退
- 数据一直源源不断的涌进
DStream
如果要用一句话来概括Spark Streaming的处理思路的话，那就是"将连续的数据持久化，离散化，然后进行批量处理"。
让我们来仔细分析一下这么作的原因。
- 数据持久化 将从网络上接收到的数据先暂时存储下来，为事件处理出错时的事件重演提供可能，
- 离散化 数据源源不断的涌进，永远没有一个尽头，就像周星驰的喜剧中所说“崇拜之情如黄河之水绵绵不绝，一发而不可收拾”。既然不能穷尽，那么就将其按时间分片。比如采用一分钟为时间间隔，那么在连续的一分钟内收集到的数据集中存储在一起。
- 批量处理 将持久化下来的数据分批进行处理，处理机制套用之前的RDD模式
DStream可以说是对RDD的又一层封装。如果打开DStream.scala和RDD.scala，可以发现几乎RDD上的所有operation在DStream中都有相应的定义。
作用于DStream上的operation分成两类
- Transformation
- Output 表示将输出结果，目前支持的有print, saveAsObjectFiles, saveAsTextFiles, saveAsHadoopFiles
DStreamGraph
有输入就要有输出，如果没有输出，则前面所做的所有动作全部没有意义，那么如何将这些输入和输出绑定起来呢？这个问题的解决就依赖于DStreamGraph，DStreamGraph记录输入的Stream和输出的Stream。
private val inputStreams = new ArrayBuffer[InputDStream[_]]()
private val outputStreams = new ArrayBuffer[DStream[_]]()
var rememberDuration: Duration = null
var checkpointInProgress = false
outputStreams中的元素是在有Output类型的Operation作用于DStream上时自动添加到DStreamGraph中的。
outputStream区别于inputStream一个重要的地方就是会重载generateJob.
初始化流程
![]()
StreamingContext
StreamingContext是Spark Streaming初始化的入口点，主要的功能是根据入参来生成JobScheduler
设定InputStream
如果流数据源来自于socket，则使用socketStream。如果数据源来自于不断变化着的文件，则可使用fileStream
提交运行
StreamingContext.start()
数据处理
![]()
以socketStream为例，数据来自于socket。
SocketInputDstream启动一个线程，该线程使用receive函数来接收数据
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)
}
logInfo("Stopped receiving")
restart("Retrying connecting to " + host + ":" + port)
} catch {
case e: java.net.ConnectException =>
restart("Error connecting to " + host + ":" + port, e)
case t: Throwable =>
restart("Error receiving data", t)
} finally {
if (socket != null) {
socket.close()
logInfo("Closed socket to " + host + ":" + port)
}
}
}
}
接收到的数据会被先存储起来，存储最终会调用到BlockManager.scala中的函数，那么BlockManager是如何被传递到StreamingContext的呢？利用SparkEnv传入的，注意StreamingContext构造函数的入参。
![]()
处理定时器
数据的存储有是被socket触发的。那么已经存储的数据被真正的处理又是被什么触发的呢？
记得在初始化StreamingContext的时候，我们指定了一个时间参数，那么用这个参数会构造相应的重复定时器，一旦定时器超时，调用generateJobs函数。
private val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds, longTime => eventActor ! GenerateJobs(new Time(longTime)), "JobGenerator")
事件处理函数
/** Processes all events */
private def processEvent(event: JobGeneratorEvent) {
logDebug("Got event " + event)
event match {
case GenerateJobs(time) => generateJobs(time)
case ClearMetadata(time) => clearMetadata(time)
case DoCheckpoint(time) => doCheckpoint(time)
case ClearCheckpointData(time) => clearCheckpointData(time)
}
}
generteJobs
private def generateJobs(time: Time) {
SparkEnv.set(ssc.env)
Try(graph.generateJobs(time)) match {
case Success(jobs) =>
val receivedBlockInfo = graph.getReceiverInputStreams.map { stream =>
val streamId = stream.id
val receivedBlockInfo = stream.getReceivedBlockInfo(time)
(streamId, receivedBlockInfo)
}.toMap
jobScheduler.submitJobSet(JobSet(time, jobs, receivedBlockInfo))
case Failure(e) =>
jobScheduler.reportError("Error generating jobs for time " + time, e)
}
eventActor ! DoCheckpoint(time)
}
generateJobs->generateJob一路下去会调用到Job.run,在job.run中调用sc.runJob，在具体调用路径就不一一列出。
private class JobHandler(job: Job) extends Runnable {
def run() {
eventActor ! JobStarted(job)
job.run()
eventActor ! JobCompleted(job)
}
}
DStream.generateJob函数中定义了jobFunc，也就是在job.run()中使用到的jobFunc
private[streaming] def generateJob(time: Time): Option[Job] = {
getOrCompute(time) match {
case Some(rdd) => {
val jobFunc = () => {
val emptyFunc = { (iterator: Iterator[T]) => {} }
context.sparkContext.runJob(rdd, emptyFunc)
}
Some(new Job(time, jobFunc))
}
case None => None
}
}
在这个流程中，DStreamGraph起到非常关键的作用，非常类似于TridentStorm中的graph.
在generateJob过程中，DStream会通过调用compute函数生成相应的RDD，SparkContext则是将基于RDD的抽象转换成为多个stage，而执行。
StreamingContext中一个重要的转换就是DStream到RDD的转换，而SparkContext中一个重要的转换是RDD到Stage及Task的转换。在这两个不同的抽象类中，要注意其中getOrCompute和compute函数的实现。
小结
本篇内容有点仓促，内容不够丰富翔实，争取回头有空的时候再好好丰富一下具体的调用路径。
对于容错处理机制，本文没有涉及，待研究明白之后另起一篇进行阐述。
<二>DStream处理的容错性分析
在流数据的处理过程中，为了保证处理结果的可信度(不能多算，也不能漏算)，需要做到对所有的输入数据有且仅有一次处理。在Spark Streaming的处理机制中，不能多算，比较容易理解。那么它又是如何作到即使数据处理结点被重启，在重启之后这些数据也会被再次处理呢？
环境搭建
为了有一个感性的认识，先运行一下简单的Spark Streaming示例。首先确认已经安装了openbsd-netcat。
运行netcat
运行spark-shell
SPARK_JAVA_OPTS=-Dspark.cleaner.ttl=10000 MASTER=local-cluster[2,2,1024] bin/spark-shell
在spark-shell中输入如下内容
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
val ssc = new StreamingContext(sc, Seconds(3))
val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap( _.split(" "))
val pairs = words.map(word => (word,1))
val wordCount = pairs.reduceByKey(_ + _)
wordCount.print()
ssc.start()
ssc.awaitTermination()
当ssc.start()执行之后，在nc一侧输入一些内容并回车，spark-shell上就会显示出统计的结果。
数据接收过程
来看一下代码实现层面，从两个角度来说，一是控制层面(control panel)，另一是数据层面(data panel)。
Spark Streaming的数据接收过程的控制层面大致如下图所示。
![]()
简要讲解一下上图的意思：
- 数据真正接收到是发生在SocketReceiver.receive函数中，将接收到的数据放入到BlockGenerator.currentBuffer
- 在BlockGenerator中有一个重复定时器，处理函数为updateCurrentBuffer, updateCurrentBuffer将当前buffer中的数据封装为一个新的Block，放入到blocksForPush队列中
- 同样是在BlockGenerator中有一个BlockPushingThread，其职责就是不停的将blocksForPush队列中的成员通过pushArrayBuffer函数传递给blockmanager,让BlockManager将数据存储到MemoryStore中
- pushArrayBuffer还会将已经由BlockManager存储的Block的id号传递给ReceiverTracker，ReceiverTracker会将存储的blockId放到对应StreamId的队列中
socket.receive->receiver.store->pushSingle->blockgenerator.updateCurrentBuffer->blockgenerator.keepPushBlocks->pushArrayBufer
->ReceiverTracker.addBlocks
pushArrayBuffer函数的定义如下
def pushArrayBuffer(
arrayBuffer: ArrayBuffer[_],
optionalMetadata: Option[Any],
optionalBlockId: Option[StreamBlockId]
) {
val blockId = optionalBlockId.getOrElse(nextBlockId)
val time = System.currentTimeMillis
blockManager.put(blockId, arrayBuffer.asInstanceOf[ArrayBuffer[Any]],
storageLevel, tellMaster = true)
logDebug("Pushed block " + blockId + " in " + (System.currentTimeMillis - time) + " ms")
reportPushedBlock(blockId, arrayBuffer.size, optionalMetadata)
}
数据结构的变化过程
Spark Streaming数据处理高效的原因之一就是批量的进行数据分析，那么这些批量的数据是如何聚集起来的呢？换种方式来表述这个问题，在某一时刻，接收到的数据是单一的，也就是我们最多只能组成<t,data>这种数据元组，而在runJob的时候是批量的提取和分析数据的，这个批量数据的组成是在什么时候完成的呢？
下图大到勾勒出一条新的message被socketreceiver接收之后，是如何通过一系列的处理而放入到BlockManager中，并同时由ReceiverTracker记录下相应的元数据的。
![]()
- 首先new message被放入到blockManager.currentBuffer
- 定时器超时处理过程，将整个currentBuffer中的数据打包成一条Block，放入到ArrayBlockingQueue,该数据结构支持FIFO
- keepPushingBlocks将每一条Block(block中包含时间戳，接收到的原始数据）让BlockManager进行保存，同时通知ReceiverTracker已经将哪些block存储到了blockmanager中
- ReceiverTracker将每一个stream接收到但还没有进行处理的block放入到receiverBlockInfo,其为一Hashmap. 在后面的generateJobs中会从receiverBlockInfo提取数据以生成相应的RDD
数据处理过程
数据处理中最重要的函数就是generateJobs, generateJobs会引发下述的函数调用过程，具体的代码就不一一罗列了。
- jobgenerator.generateJobs->dstreamgraph.generateJobs->dstream.generateJob->getOrCompute->compute 生成RDD
- job调用job.func
JobGenerator.generateJobs函数定义如下
private def generateJobs(time: Time) {
SparkEnv.set(ssc.env)
Try(graph.generateJobs(time)) match {
case Success(jobs) =>
val receivedBlockInfo = graph.getReceiverInputStreams.map { stream =>
val streamId = stream.id
val receivedBlockInfo = stream.getReceivedBlockInfo(time)
(streamId, receivedBlockInfo)
}.toMap
jobScheduler.submitJobSet(JobSet(time, jobs, receivedBlockInfo))
case Failure(e) =>
jobScheduler.reportError("Error generating jobs for time " + time, e)
}
eventActor ! DoCheckpoint(time)
}
我们先来谈一谈数据处理阶段是如何与上述的接收阶段中存储下来的数据挂上钩的。
假设上一次进行RDD处理发生在时间点t1,现在是时间点t2，那么在<t2,t1>之间有哪些blocks没有被处理呢？
想必你已经知道答案了，没有被处理的blocks全部保存在ReceiverTracker的receiverBlockInfo之中
在generateJob时，每一个DStream都会调用getReceivedBlockInfo，你说没有跟ReceiverTracker中的receivedBlockInfo连起来啊，别急！且看数据输入的源头ReceiverInputDStream中的getReceivedBlockInfo是如何定义的。代码列举如下。
private[streaming] def getReceivedBlockInfo(time: Time) = {
receivedBlockInfo(time)
}
那么此处的receivedBlockInfo(time)是从何而来的呢，这个要看ReceivedInputDStream中的compute函数实现
override def compute(validTime: Time): Option[RDD[T]] = {
if (validTime >= graph.startTime) {
val blockInfo = ssc.scheduler.receiverTracker.getReceivedBlockInfo(id)
receivedBlockInfo(validTime) = blockInfo
val blockIds = blockInfo.map(_.blockId.asInstanceOf[BlockId])
Some(new BlockRDD[T](ssc.sc, blockIds))
} else {
Some(new BlockRDD[T](ssc.sc, Array[BlockId]()))
}
}
至此终于看到了receiverTracker中的getReceivedBlockInfo被调用，也就是说将接收阶段的数据和目前处理阶段的输入通道打通了
函数调用路径，从generateJobs到sparkcontext.submitJobs. 这个时候要注意注册为DStreamGraph中的OutputStream上的操作会引发SparkContext.runJobs被调用。我们以print函数为例看一下调用过程。
def print() {
def foreachFunc = (rdd: RDD[T], time: Time) => {
val first11 = rdd.take(11)
println ("-------------------------------------------")
println ("Time: " + time)
println ("-------------------------------------------")
first11.take(10).foreach(println)
if (first11.size > 10) println("...")
println()
}
new ForEachDStream(this, context.sparkContext.clean(foreachFunc)).register()
}
注意rdd.take，这个会引发runJob调用,不信的话，我们可以看一看其定义中调用runJob的片段。
val left = num - buf.size
val p = partsScanned until math.min(partsScanned + numPartsToTry, totalParts)
val res = sc.runJob(this, (it: Iterator[T]) => it.take(left).toArray, p, allowLocal = true)
res.foreach(buf ++= _.take(num - buf.size))
partsScanned += numPartsToTry
}
小结一下数据处理过程
- 用time为关键字去取出在此时间之前加入的所有blockIds
- 真正提交运行的时候，rdd中的blockfetcher以blockId为关键字去blockmanagermaster获取真正的数据，即从socket上接收到的原始数据
容错处理
JobGenerator.generateJobs函数的最后会发出DoCheckpoint通知，该通知会让相应的actor将DStreamCheckpointData写入到hdfs文件中，我们来看一看为什么需要写入checkpointdata以及哪些东西是包含在checkpointdata之中。
![]()
在数据处理一节，我们已经分析到在generateJobs的时候会生成多个jobs,它们会通过sparkcontext.runJob接口而发送到cluster中被真正执行。
假设在t2，worker挂掉了，挂掉的worker直到t3才完全恢复。由于挂掉的原因，上一次generateJobs生成的job不一定被完全处理了(也许有些已经处理了，有些还没有处理)，所以需要重新再提交一次。这里有一个问题，那就是可能导致针对同一批数据有重复处理的情况发生，从而无法达到exactly-once的语义效果。
问题2： 在<t2,t3>这一段挂掉的时间之内，没有新的数据被接收，所以Spark Streaming的SocketReceiver适合用来充当client侧而不是server侧。SocketReceiver读取到的数据应该存在一个具有冗余备份机制的内存数据库或缓存队列里，如kafaka. 对问题2, Spark Streaming本身是解决不了的。当然这里再往下细究的话，会牵出负载均衡的问题。
checkpointData
checkpoint的成员变量有哪些呢，我们看一看其结构定义就清楚了。
val master = ssc.sc.master
val framework = ssc.sc.appName
val sparkHome = ssc.sc.getSparkHome.getOrElse(null)
val jars = ssc.sc.jars
val graph = ssc.graph
val checkpointDir = ssc.checkpointDir
val checkpointDuration = ssc.checkpointDuration
val pendingTimes = ssc.scheduler.getPendingTimes().toArray
val delaySeconds = MetadataCleaner.getDelaySeconds(ssc.conf)
val sparkConfPairs = ssc.conf.getAll
generatedRDDs是被包含在graph里面。所以不要突然之间惊惶失措，发觉没有将generatedRDDs保存起来。
checkpoint的数据是通过CheckpointwriteHandler真正的写入到hdfs，通过CheckPiontReader而读入。CheckpointReade在重启的时候会被使用到，判断是第一次干净的启动还是因错误而重启，判断的依据全部在cp这个变量。
为了达到重启之后而自动的检查并载入相应的checkpoint数据，那么在创建StreamingContext的时候就不能简单的通过调用new StreamingContext来完成，而是利用getOrCreate函数，代码示例如下。
def functionToCreateContext(): StreamingContext = {
val ssc = new StreamingContext(...)
val lines = ssc.socketTextStream(...)
...
ssc.checkpoint(checkpointDirectory)
ssc
}
val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)
context. ...
context.start()
context.awaitTermination()
小结
本文中讲述数据接收过程中所使用的两幅图使用tikz完成，里面包含的信息很丰富，有志于了解清楚Spark Streaming内部处理机制的同仁，不妨以此为参考进行详细的代码走读。
如果有任何不对或错误之处，欢迎批评指正。
参考资料
- Spark Streaming源码分析 checkpoint http://www.cnblogs.com/fxjwind/p/3596451.html
- Spark Streaming Introduction http://jerryshao.me/architecture/2013/04/02/spark-streaming-introduction/
- deep dive with Spark Streaming http://www.meetup.com/spark-users/events/122694912/
