######好好好#######DStream 生成 RDD 实例详解

// ssc.socketTextStream() 将创建一个 SocketInputDStream；这个 InputDStream 的 SocketReceiver 将监听本机 9999 端口
val lines = ssc.socketTextStream("localhost", 9999)val words = lines.flatMap(_.split(" "))      // DStream transformation
val pairs = words.map(word => (word, 1))     // DStream transformation
val wordCounts = pairs.reduceByKey(_ + _)    // DStream transformation
wordCounts.print()                           // DStream output

def socketStream[T: ClassTag](hostname: String, port: Int, converter: (InputStream) => Iterator[T], storageLevel: StorageLevel): ReceiverInputDStream[T] = {new SocketInputDStream[T](this, hostname, port, converter, storageLevel)
}

def flatMap[U: ClassTag](flatMapFunc: T => Traversable[U]): DStream[U] = ssc.withScope {new FlatMappedDStream(this, context.sparkContext.clean(flatMapFunc))
}

val lines = new SocketInputDStream("localhost", 9999)   // 类型是 SocketInputDStreamval words = new FlatMappedDStream(lines, _.split(" "))  // 类型是 FlatMappedDStream
val pairs = new MappedDStream(words, word => (word, 1)) // 类型是 MappedDStream
val wordCounts = new ShuffledDStream(pairs, _ + _)      // 类型是 ShuffledDStream
new ForeachDStream(wordCounts, cnt => cnt.print())      // 类型是 ForeachDStream

private[streaming] var generatedRDDs = new HashMap[Time, RDD[T]] ()

private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = {// 从 generatedRDDs 里 get 一下：如果有 rdd 就返回，没有 rdd 就进行 orElse 下面的 rdd 生成步骤generatedRDDs.get(time).orElse {// 验证 time 需要是 validif (isTimeValid(time)) {// 然后调用 compute(time) 方法获得 rdd 实例，并存入 rddOption 变量val rddOption = createRDDWithLocalProperties(time) {PairRDDFunctions.disableOutputSpecValidation.withValue(true) {compute(time)}}rddOption.foreach { case newRDD =>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")}// 将刚刚实例化出来的 rddOption 放入 generatedRDDs 对应的 time 位置generatedRDDs.put(time, newRDD)}// 返回刚刚实例化出来的 rddOptionrddOption} else {None}}}

// 来自 FileInputDStream
override def compute(validTime: Time): Option[RDD[(K, V)]] = {// 通过一个 findNewFiles() 方法，找到 validTime 以后产生的新 file 的数据val newFiles = findNewFiles(validTime.milliseconds)logInfo("New files at time " + validTime + ":\n" + newFiles.mkString("\n"))batchTimeToSelectedFiles += ((validTime, newFiles))recentlySelectedFiles ++= newFiles// 找到了一些新 file；以新 file 的数组为参数，通过 filesToRDD() 生成单个 RDD 实例 rddsval rdds = Some(filesToRDD(newFiles))val metadata = Map("files" -> newFiles.toList,StreamInputInfo.METADATA_KEY_DESCRIPTION -> newFiles.mkString("\n"))val inputInfo = StreamInputInfo(id, 0, metadata)ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)// 返回生成的单个 RDD 实例 rddsrdds}

// 来自 FileInputDStream
private def filesToRDD(files: Seq[String]): RDD[(K, V)] = {// 对每个 file，都 sc.newAPIHadoopFile(file) 来生成一个 RDDval fileRDDs = files.map { file =>val rdd = serializableConfOpt.map(_.value) match {case Some(config) => context.sparkContext.newAPIHadoopFile(file,fm.runtimeClass.asInstanceOf[Class[F]],km.runtimeClass.asInstanceOf[Class[K]],vm.runtimeClass.asInstanceOf[Class[V]],config)case None => context.sparkContext.newAPIHadoopFile[K, V, F](file)}if (rdd.partitions.size == 0) {logError("File " + file + " has no data in it. Spark Streaming can only ingest " +"files that have been \"moved\" to the directory assigned to the file stream. " +"Refer to the streaming programming guide for more details.")}rdd}// 将每个 file 对应的 RDD 进行 union，返回一个 union 后的 UnionRDDnew UnionRDD(context.sparkContext, fileRDDs)
}

package org.apache.spark.streaming.dstreamimport org.apache.spark.streaming.{Duration, Time}
import org.apache.spark.rdd.RDD
import scala.reflect.ClassTagprivate[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.slideDurationoverride def compute(validTime: Time): Option[RDD[U]] = {parent.getOrCompute(validTime).map(_.map[U](mapFunc))}
}

package org.apache.spark.streaming.dstreamimport org.apache.spark.streaming.{Duration, Time}
import org.apache.spark.rdd.RDD
import scala.reflect.ClassTagprivate[streaming]
class FilteredDStream[T: ClassTag](parent: DStream[T],filterFunc: T => Boolean) extends DStream[T](parent.ssc) {override def dependencies: List[DStream[_]] = List(parent)override def slideDuration: Duration = parent.slideDurationoverride def compute(validTime: Time): Option[RDD[T]] = {parent.getOrCompute(validTime).map(_.filter(filterFunc))}
}

package org.apache.spark.streaming.dstreamimport org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Duration, Time}
import org.apache.spark.streaming.scheduler.Job
import scala.reflect.ClassTagprivate[streaming]
class ForEachDStream[T: ClassTag] (parent: DStream[T],foreachFunc: (RDD[T], Time) => Unit) extends DStream[Unit](parent.ssc) {override def dependencies: List[DStream[_]] = List(parent)override def slideDuration: Duration = parent.slideDurationoverride def compute(validTime: Time): Option[RDD[Unit]] = Noneoverride def generateJob(time: Time): Option[Job] = {parent.getOrCompute(time) match {case Some(rdd) =>val jobFunc = () => createRDDWithLocalProperties(time) {ssc.sparkContext.setCallSite(creationSite)foreachFunc(rdd, time)}Some(new Job(time, jobFunc))case None => None}}
}

def foreachFunc: (RDD[T], Time) => Unit = {val firstNum = rdd.take(num + 1)println("-------------------------------------------")println("Time: " + time)println("-------------------------------------------")firstNum.take(num).foreach(println)if (firstNum.length > num) println("...")println()
}

// 来自 DStreamGraph
def generateJobs(time: Time): Seq[Job] = {logDebug("Generating jobs for time " + time)val jobs = this.synchronized {outputStreams.flatMap(outputStream => outputStream.generateJob(time))}logDebug("Generated " + jobs.length + " jobs for time " + time)jobs
}

// 节选自 org.apache.spark.streaming.scheduler.Job
private[streaming]
class Job(val time: Time, func: () => _) {...def run() {_result = Try(func())}...
}

// 来自 ForEachDStream
override def generateJob(time: Time): Option[Job] = {// 【首先调用 parentDStream 的 getOrCompute() 来获取 parentRDD】parent.getOrCompute(time) match {case Some(rdd) =>// 【然后定义 jobFunc 为在 parentRDD 上执行 foreachFun() 】val jobFunc = () => createRDDWithLocalProperties(time) {ssc.sparkContext.setCallSite(creationSite)foreachFunc(rdd, time)}// 【最后将 jobFunc 包装为 Job 返回】Some(new Job(time, jobFunc))case None => None}
}