本期内容：微信
1，DStream与RDD关系完全研究app
2，Streaming中RDD的生成完全研究框架
RDD是怎么生成的？RDD依靠什么生成？RDD生成的依据是什么？Spark Streaming中RDD的执行是否和Spark Core中的RDD执行有所不一样？运行以后咱们对RDD怎么处理？socket
RDD自己也是基本的对象，例如说BatchInterval为1秒，那么每一秒都会产生RDD，内存中不能彻底容纳该对象。每一个BatchInterval的做业执行完后，怎么对已有的RDD进行管理。ide
ForEachDStream不必定会触发Job的执行，和Job的执行没有关系。函数
Job的产生是由Spark Streaming框架形成的。oop
foreachRDD是Spark Streaming的后门，能够直接对RDD进行操做。post
DStream就是RDD的模板，后面的DStream与前面的DStream有依赖。大数据
val lines = jsc.socketTextStream("127.0.0.1", 9999)这里产生了SocketInputDStream。ui
lines.flatMap(_.split(" ")).map(word => (word, 1)).reduceByKey(_ + _).print()这里由SocketInputDStream转换为FlatMappedDStream，再转换为MappedDStream，再转换为ShuffledDStream，再转换为ForEachDStream。
对于DStream类，源码中是这样解释的。
| * DStreams internally is characterized by a few basic properties:
* - A list of other DStreams that the DStream depends on
* - A time interval at which the DStream generates an RDD
* - A function that is used to generate an RDD after each time interval |
大体意思是：
1.DStream依赖于其余DStream。
2.每隔BatchDuration，DStream生成一个RDD
3.每隔BatchDuration，DStream内部函数会生成RDD
DStream是从后往前依赖，由于DStream表明Spark Streaming业务逻辑，RDD是从后往前依赖的，DStream是lazy级别的。DStream的依赖关系必须和RDD的依赖关系保持高度一致。
DStream类中generatedRDDs存储着不一样时间对应的RDD实例。每个DStream实例都有本身的generatedRDDs。实际运算的时候，因为是从后往前推，计算只做用于最后一个DStream。
| // RDDs generated, marked as private[streaming] so that testsuites can access it
@transient
private[streaming] var generatedRDDs = new HashMap[Time, RDD[T]] () |
generatedRDDs是如何获取的。DStream的getOrCompute方法，先根据时间判断HashMap中是否已存在该时间对应的RDD，若是没有则调用compute获得RDD，并放入到HashMap中。
| /**
* Get the RDD corresponding to the given time; either retrieve it from cache
* or compute-and-cache it.
*/
private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = {
// If RDD was already generated, then retrieve it from HashMap,
// or else compute the RDD
generatedRDDs.get(time).orElse {
// Compute the RDD if time is valid (e.g. correct time in a sliding window)
// of RDD generation, else generate nothing.
if (isTimeValid(time)) {
val rddOption = createRDDWithLocalProperties(time, displayInnerRDDOps = false) {
// Disable checks for existing output directories in jobs launched by the streaming
// scheduler, since we may need to write output to an existing directory during checkpoint
// recovery; see SPARK-4835 for more details. We need to have this call here because
// compute() might cause Spark jobs to be launched.
PairRDDFunctions.disableOutputSpecValidation.withValue(true) {
compute(time)
}
}
rddOption.foreach { case newRDD =>
// Register the generated RDD for caching and checkpointing
if (storageLevel != StorageLevel.NONE) {
newRDD.persist(storageLevel)
logDebug(s"Persisting RDD ${newRDD.id} for time $time to $storageLevel")
}
if (checkpointDuration != null && (time - zeroTime).isMultipleOf(checkpointDuration)) {
newRDD.checkpoint()
logInfo(s"Marking RDD ${newRDD.id} for time $time for checkpointing")
}
generatedRDDs.put(time, newRDD)
}
rddOption
} else {
None
}
}
} |
拿DStream的子类ReceiverInputDStream来讲明compute方法，内部调用了createBlockRDD这个方法。
| /**
* Generates RDDs with blocks received by the receiver of this stream. */
override def compute(validTime: Time): Option[RDD[T]] = {
val blockRDD = {
if (validTime < graph.startTime) {
// If this is called for any time before the start time of the context,
// then this returns an empty RDD. This may happen when recovering from a
// driver failure without any write ahead log to recover pre-failure data.
new BlockRDD[T](ssc.sc, Array.empty)
} else {
// Otherwise, ask the tracker for all the blocks that have been allocated to this stream
// for this batch
val receiverTracker = ssc.scheduler.receiverTracker
val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)
// Register the input blocks information into InputInfoTracker
val inputInfo = StreamInputInfo(id, blockInfos.flatMap(_.numRecords).sum)
ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)
// Create the BlockRDD
createBlockRDD(validTime, blockInfos)
}
}
Some(blockRDD)
} |
createBlockRDD会返回BlockRDD，因为ReceiverInputDStream没有父依赖，因此本身生成RDD。
| private[streaming] def createBlockRDD(time: Time, blockInfos: Seq[ReceivedBlockInfo]): RDD[T] = {
if (blockInfos.nonEmpty) {
val blockIds = blockInfos.map { _.blockId.asInstanceOf[BlockId] }.toArray
// Are WAL record handles present with all the blocks
val areWALRecordHandlesPresent = blockInfos.forall { _.walRecordHandleOption.nonEmpty }
if (areWALRecordHandlesPresent) {
// If all the blocks have WAL record handle, then create a WALBackedBlockRDD
val isBlockIdValid = blockInfos.map { _.isBlockIdValid() }.toArray
val walRecordHandles = blockInfos.map { _.walRecordHandleOption.get }.toArray
new WriteAheadLogBackedBlockRDD[T](
ssc.sparkContext, blockIds, walRecordHandles, isBlockIdValid)
} else {
// Else, create a BlockRDD. However, if there are some blocks with WAL info but not
// others then that is unexpected and log a warning accordingly.
if (blockInfos.find(_.walRecordHandleOption.nonEmpty).nonEmpty) {
if (WriteAheadLogUtils.enableReceiverLog(ssc.conf)) {
logError("Some blocks do not have Write Ahead Log information; " +
"this is unexpected and data may not be recoverable after driver failures")
} else {
logWarning("Some blocks have Write Ahead Log information; this is unexpected")
}
}
val validBlockIds = blockIds.filter { id =>
ssc.sparkContext.env.blockManager.master.contains(id)
}
if (validBlockIds.size != blockIds.size) {
logWarning("Some blocks could not be recovered as they were not found in memory. " +
"To prevent such data loss, enabled Write Ahead Log (see programming guide " +
"for more details.")
}
new BlockRDD[T](ssc.sc, validBlockIds)
}
} else {
// If no block is ready now, creating WriteAheadLogBackedBlockRDD or BlockRDD
// according to the configuration
if (WriteAheadLogUtils.enableReceiverLog(ssc.conf)) {
new WriteAheadLogBackedBlockRDD[T](
ssc.sparkContext, Array.empty, Array.empty, Array.empty)
} else {
new BlockRDD[T](ssc.sc, Array.empty)
}
}
} |
再拿DStream的子类MappedDStream来讲，这里的compute方法，是调用父RDD的getOrCompute方法得到RDD，再使用map操做。
| private[streaming]
class MappedDStream[T: ClassTag, U: ClassTag] (
parent: DStream[T],
mapFunc: T => U
) extends DStream[U](parent.ssc) {
override def dependencies: List[DStream[_]] = List(parent)
override def slideDuration: Duration = parent.slideDuration
override def compute(validTime: Time): Option[RDD[U]] = {
parent.getOrCompute(validTime).map(_.map[U](mapFunc))
}
} |
从上面两个DStream的子类，能够说明第一个DStream，即InputDStream的comput方法是本身获取数据并计算的，而其余的DStream是依赖父DStream的，调用父DStream的getOrCompute方法，而后进行计算。
以上说明了对DStream的操做最后做用于对RDD的操做。
接着看下DStream的另外一个子类ForEachDStream，发现其compute方法没有任何操做，可是重写了generateJob方法。
| private[streaming]
class ForEachDStream[T: ClassTag] (
parent: DStream[T],
foreachFunc: (RDD[T], Time) => Unit,
displayInnerRDDOps: Boolean
) extends DStream[Unit](parent.ssc) {
override def dependencies: List[DStream[_]] = List(parent)
override def slideDuration: Duration = parent.slideDuration
override def compute(validTime: Time): Option[RDD[Unit]] = None
override def generateJob(time: Time): Option[Job] = {
parent.getOrCompute(time) match {
case Some(rdd) =>
val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
foreachFunc(rdd, time)
}
Some(new Job(time, jobFunc))
case None => None
}
}
} |
从Job生成入手，JobGenerator的generateJobs方法，内部调用的DStreamGraph的generateJobs方法。
| /** Generate jobs and perform checkpoint for the given `time`. */
private def generateJobs(time: Time) {
// Set the SparkEnv in this thread, so that job generation code can access the environment
// Example: BlockRDDs are created in this thread, and it needs to access BlockManager
// Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed.
SparkEnv.set(ssc.env)
Try {
//根据特定的时间获取具体的数据
jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch
//调用DStreamGraph的generateJobs生成Job
graph.generateJobs(time) // generate jobs using allocated block
} 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)
}
eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
} |
DStreamGraph的generateJobs方法调用了OutputStream的generateJob方法，OutputStream就是ForEachDStream。
| def generateJobs(time: Time): Seq[Job] = {
logDebug("Generating jobs for time " + time)
val jobs = this.synchronized {
outputStreams.flatMap { outputStream =>
val jobOption = outputStream.generateJob(time)
jobOption.foreach(_.setCallSite(outputStream.creationSite))
jobOption
}
}
logDebug("Generated " + jobs.length + " jobs for time " + time)
jobs
} |
总结：DStream是RDD的模板，其内部generatedRDDs 保存了每一个BatchDuration时间生成的RDD对象实例。DStream的依赖构成了RDD依赖关系，即从后往前计算时，只要对最后一个DStream计算便可。JobGenerator每隔BatchDuration调用DStreamGraph的generateJobs方法，调用了ForEachDStream的generateJob方法，其内部先调用父DStream的getOrCompute方法来获取RDD，而后在进行计算，从后往前推，第一个DStream是ReceiverInputDStream，其comput方法中从receiverTracker中获取对应时间段的metadata信息，而后生成BlockRDD对象，并放入到generatedRDDs中。
备注：
一、DT大数据梦工厂微信公众号DT_Spark 二、IMF晚8点大数据实战YY直播频道号：68917580 三、新浪微博: http://www.weibo.com/ilovepains