第5课：基于案例一节课贯通Spark Streaming流计算框架的运行源码

第5课：基于案例一节课贯通Spark Streaming流计算框架的运行源码

这一节课基于一个案例贯通sparkstreaming的源码。

本课内容：

1 在线动态计算分类最热门商品案例回顾与演示

2 基于案例贯通Spark Streaming的运行源码

　　一切不能进行实时流处理的数据都是无效的数据。在流处理时代，SparkStreaming有着强大吸引力，而且发展前景广阔，加之Spark的生态系统，Streaming可以方便调用其他的诸如SQL，MLlib等强大框架，它必将一统天下。

Spark Streaming运行时与其说是Spark Core上的一个流式处理框架，不如说是Spark Core上的一个最复杂的应用程序。如果可以掌握Spark
streaming这个复杂的应用程序，那么其他的再复杂的应用程序都不在话下了。这里选择Spark Streaming作为版本定制的切入点也是大势所趋。

package com.dt.spark.sparkstreaming

import com.robinspark.utils.ConnectionPool

import org.apache.spark.SparkConf

import org.apache.spark.sql.Row

import org.apache.spark.sql.hive.HiveContext

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

import org.apache.spark.streaming.{Seconds, StreamingContext}

/**

* 使用Spark Streaming+Spark SQL来在线动态计算电商中不同类别中最热门的商品排名，例如手机这个类别下面最热门的三种手机、电视这个类别

* 下最热门的三种电视，该实例在实际生产环境下具有非常重大的意义；

*

* @author DT大数据梦工厂

* 新浪微博：http://weibo.com/ilovepains/

*

*

* 实现技术：Spark Streaming+Spark SQL，之所以Spark Streaming能够使用ML、sql、graphx等功能是因为有foreachRDD和Transform

* 等接口，这些接口中其实是基于RDD进行操作，所以以RDD为基石，就可以直接使用Spark其它所有的功能，就像直接调用API一样简单。

* 假设说这里的数据的格式：user item category，例如Rocky Samsung Android

*/

object OnlineTheTop3ItemForEachCategory2DB {

def main(args: Array[String]){

/**

* 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，

*/

val conf = new SparkConf() //创建SparkConf对象

conf.setAppName("OnlineTheTop3ItemForEachCategory2DB") //设置应用程序的名称，在程序运行的监控界面可以看到名称

conf.setMaster("spark://Master:7077") //此时，程序在Spark集群

//conf.setMaster("local[2]")

//设置batchDuration时间间隔来控制Job生成的频率并且创建Spark
Streaming执行的入口

val ssc = new StreamingContext(conf, Seconds(5))

ssc.checkpoint("/root/Documents/SparkApps/checkpoint")

val userClickLogsDStream = ssc.socketTextStream("Master", 9999)

val formattedUserClickLogsDStream = userClickLogsDStream.map(clickLog =>

(clickLog.split(" ")(2) + "_" + clickLog.split(" ")(1), 1))

// val categoryUserClickLogsDStream = formattedUserClickLogsDStream.reduceByKeyAndWindow((v1:Int, v2: Int) => v1 + v2,

// (v1:Int, v2: Int) => v1 - v2, Seconds(60), Seconds(20))

val categoryUserClickLogsDStream = formattedUserClickLogsDStream.reduceByKeyAndWindow(_+_,

_-_, Seconds(60), Seconds(20))

categoryUserClickLogsDStream.foreachRDD { rdd => {

if (rdd.isEmpty()) {

println("No data inputted!!!")

} else {

val categoryItemRow = rdd.map(reducedItem => {

val category = reducedItem._1.split("_")(0)

val item = reducedItem._1.split("_")(1)

val click_count = reducedItem._2

Row(category, item, click_count)

})

val structType = StructType(Array(

StructField("category", StringType, true),

StructField("item", StringType, true),

StructField("click_count", IntegerType, true)

))

val hiveContext = new HiveContext(rdd.context)

val categoryItemDF = hiveContext.createDataFrame(categoryItemRow, structType)

categoryItemDF.registerTempTable("categoryItemTable")

val reseltDataFram = hiveContext.sql("SELECT category,item,click_count FROM (SELECT category,item,click_count,row_number()" +

" OVER (PARTITION BY category ORDER BY click_count DESC) rank FROM categoryItemTable) subquery " +

" WHERE rank <= 3")

reseltDataFram.show()

val resultRowRDD = reseltDataFram.rdd

resultRowRDD.foreachPartition { partitionOfRecords => {

if (partitionOfRecords.isEmpty){

println("This RDD is not null but partition is null")

} else {

// ConnectionPool is a static, lazily initialized pool of connections

val connection = ConnectionPool.getConnection()

partitionOfRecords.foreach(record => {

val sql = "insert into categorytop3(category,item,client_count) values('" + record.getAs("category") + "','" +

record.getAs("item") + "'," + record.getAs("click_count") + ")"

val stmt = connection.createStatement();

stmt.executeUpdate(sql);

})

ConnectionPool.returnConnection(connection) // return to the pool for future reuse

}

}

}

}

}

}

/**

* 在StreamingContext调用start方法的内部其实是会启动JobScheduler的Start方法，进行消息循环，在JobScheduler

* 的start内部会构造JobGenerator和ReceiverTacker，并且调用JobGenerator和ReceiverTacker的start方法：

* 1，JobGenerator启动后会不断的根据batchDuration生成一个个的Job

* 2，ReceiverTracker启动后首先在Spark Cluster中启动Receiver（其实是在Executor中先启动ReceiverSupervisor），在Receiver收到

* 数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker，在ReceiverTracker

* 内部会通过ReceivedBlockTracker来管理接受到的元数据信息

* 每个BatchInterval会产生一个具体的Job，其实这里的Job不是Spark
Core中所指的Job，它只是基于DStreamGraph而生成的RDD

* 的DAG而已，从Java角度讲，相当于Runnable接口实例，此时要想运行Job需要提交给JobScheduler，在JobScheduler中通过线程池的方式找到一个

* 单独的线程来提交Job到集群运行（其实是在线程中基于RDD的Action触发真正的作业的运行），为什么使用线程池呢？

* 1，作业不断生成，所以为了提升效率，我们需要线程池；这和在Executor中通过线程池执行Task有异曲同工之妙；

* 2，有可能设置了Job的FAIR公平调度的方式，这个时候也需要多线程的支持；

*

*/

ssc.start()

ssc.awaitTermination()

}

}

下面就基于源码进行分析：

/**

* Create a StreamingContext by providing the details necessary for creating a new SparkContext.

* @param master cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4]).

* @param appName a name for your job, to display on the cluster web UI

* @param batchDuration the time interval at which streaming data will be divided into batches

*/

def this(

master: String,

appName: String,

batchDuration: Duration,

sparkHome: String = null,

jars: Seq[String] = Nil,

environment: Map[String, String] = Map()) = {

this(StreamingContext.createNewSparkContext(master, appName, sparkHome, jars, environment),

null, batchDuration)

}

由此可见，StreamingContext 在内部会基于sparkconf构建出sparkcontext。

private[streaming] def createNewSparkContext(conf: SparkConf): SparkContext = {

new SparkContext(conf)

}

创建socket获取输入流

def socketTextStream(

hostname: String,

port: Int,

storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2

): ReceiverInputDStream[String] = withNamedScope("socket text stream") {

socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)

}

创建自己的Receiver来接收数据

private[streaming]

class SocketInputDStream[T: ClassTag](

ssc_ : StreamingContext,

host: String,

port: Int,

bytesToObjects: InputStream => Iterator[T],

storageLevel: StorageLevel

) extends ReceiverInputDStream[T](ssc_) {

def getReceiver(): Receiver[T] = {

new SocketReceiver(host, port, bytesToObjects, storageLevel)

}

}

Receiver通过receive方法来接收数据

extends extends

Extends

ForEachDStream代表了Dstream的输出操作

/**

* An internal DStream used to represent output operations like DStream.foreachRDD.

* @param parent Parent DStream

* @param foreachFunc Function to apply on each RDD generated by the parent DStream

* @param displayInnerRDDOps Whether the detailed callsites and scopes of the RDDs generated

* by `foreachFunc` will be displayed in the UI; only the scope and

* callsite of `DStream.foreachRDD` will be displayed.

*/

private[streaming]

class ForEachDStream[T: ClassTag] (

parent: DStream[T],

foreachFunc: (RDD[T], Time) => Unit,

displayInnerRDDOps: Boolean

) extends DStream[Unit](parent.ssc) {

ForEachDStream中有generateJob方法，虽然job看起来是在JobGenerator中产生的，但是还是调的ForEachDStream中的generateJob方法。

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

}

}

生成RDD

/**

* 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

}

}

}

程序开始运行，在一个新的线程中启动JobScheduler

/**

* Start the execution of the streams.

*

* @throws IllegalStateException if the StreamingContext is already stopped.

*/

def start(): Unit = synchronized {

state match {

case INITIALIZED =>

startSite.set(DStream.getCreationSite())

StreamingContext.ACTIVATION_LOCK.synchronized {

StreamingContext.assertNoOtherContextIsActive()

try {

validate()

// Start the streaming scheduler in a new thread, so that thread local properties

// like call sites and job groups can be reset without affecting those of the

// current thread.

ThreadUtils.runInNewThread("streaming-start") {

sparkContext.setCallSite(startSite.get)

sparkContext.clearJobGroup()

sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false")

scheduler.start()

}

state = StreamingContextState.ACTIVE

} catch {

case NonFatal(e) =>

logError("Error starting the context, marking it as stopped", e)

scheduler.stop(false)

state = StreamingContextState.STOPPED

throw e

}

StreamingContext.setActiveContext(this)

}

shutdownHookRef = ShutdownHookManager.addShutdownHook(

StreamingContext.SHUTDOWN_HOOK_PRIORITY)(stopOnShutdown)

// Registering Streaming Metrics at the start of the StreamingContext

assert(env.metricsSystem != null)

env.metricsSystem.registerSource(streamingSource)

uiTab.foreach(_.attach())

logInfo("StreamingContext started")

case ACTIVE =>

logWarning("StreamingContext has already been started")

case STOPPED =>

throw new IllegalStateException("StreamingContext has already been stopped")

}

}

通过给自己内部类发消息的方式调用启动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)

}