Dream_Spark版本定制第一课

从今天起,我们踏上了新的Spark学习旅途。我们的目标是要像Spark官方机构那样有能力去定制Spark。

一.  我们最开始将从Spark Streaming入手。

为何从Spark Streaming切入Spark定制?Spark的子框架已有若干,为何选择Spark Streaming?让我们细细道来。

1.  Spark最开始只有Spark Core,没有目前的这些子框架。这些子框架是构建于Spark Core之上的。没有哪个子框架能摆脱Spark Core。我们通过对一个框架的彻底研究,肯定可以领会Spark力量的源泉,并精通所有问题的解决之道。

2.  我们再看看目前的这些子框架。Spark SQL有太多语法,研究这些会太浪费精力。SparkR还没完善。Spark GraphX已无太多可改进之处,图计算相关的数学知识也不是目前重点。Spark MLlib中的机器学习也有太多算法是与数学相关,也不是做改进的好的选择 。所以我们选择了Spark Streaming。

二 .对Spark Streaming的理解

1. Spark Streaming是流式计算,当今时代是一个流处理时代,一切数据如果不是流式处理, 或者说和流式处理不相关的话,都是无效的数据。

2. 流式处理才是我们对大数据的初步印象,而不是批处理和数据挖掘,当然Spark强悍的地方在于,他的流式处理可以在线的直接使用机器学习、图计算、SparkSQL、Spark R的成果。

3. 整个Spark的程序,基于Spark Streaming的最容易出问题,也是最受关注的地方,也是最需要人才的地方。

4. Spark Streaming和其他子框架的不同之处,Spark Streaming很像基于Spark Core之上的应用程序。

5. 寻龙点穴,Spark就是龙脉,Spark Streaming就是穴位

2015年是流式处理的一年。大家考虑用Spark,主要也是因为Spark Streaming。这是一个流处理的时代,一切数据如果与流式处理不相关的话,都是无效的数据。Spark之所以强悍的一个重要原因在于,它的流式处理可以在线使用图计算、机器学习或者SparkR的成果,这得益于Spark一体化、多元化的基础架构设计。也就是在Spark Streaming中可以调用其它子框架,无需任何设置。这是Spark的无可匹敌之处,也是Spark Streaming必将一统天下的根源。但Spark的应用中,Spark Streaming也是最容易出问题的。

Spark Streaming与其它子框架不同之处在于,它更像是Spark Core之上的一个应用程序。所以如果要做Spark的定制开发,Spark Streaming则提供了最好的参考。你想掌握Spark Streaming,但你不去精通Spark Core的话,那是不可能的。所以我们选择Spark Streaming来提升自己,是找到了关键点。

本期内容

1 Spark Streaming另类在线实验

2 瞬间理解Spark Streaming本质

1 Spark Streaming另类在线实验

我们在研究Spark Streaming的过程中,会有困惑的事情:如何清晰的看到数据的流入、被处理的过程?

使用一个小技巧,通过调节放大Batch Interval的方式,来降低批处理次数,以方便看清楚各个环节。

我们从已写过的广告点击的在线黑名单过滤的Spark Streaming应用程序入手。

ackage com.dt.spark.streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.streaming.Seconds

object OnlineBlackListFilter {

def main(args: Array[String]){

/**

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

* 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置

* 为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如

* 只有1G的内存)的初学者。

*/

// 创建SparkConf对象

val conf = new SparkConf()

// 设置应用程序的名称,在程序运行的监控界面可以看到名称

conf.setAppName("OnlineBlackListFilter")

// 此时,程序在Spark集群

conf.setMaster("spark://Master:7077")

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

/**

* 黑名单数据准备,实际上黑名单一般都是动态的,例如在Redis或者数据库中,

* 黑名单的生成往往有复杂的业务逻辑,具体情况算法不同,

* 但是在Spark Streaming进行处理的时候每次都能够访问完整的信息。

*/

val blackList = Array(("Spy", true),("Cheater", true))

val blackListRDD = ssc.sparkContext.parallelize(blackList, 8)

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

/**

* 此处模拟的广告点击的每条数据的格式为:time、name

* 此处map操作的结果是name、(time,name)的格式

*/

val adsClickStreamFormatted = adsClickStream.map { ads => (ads.split(" ")(1), ads) }

adsClickStreamFormatted.transform(userClickRDD => {

// 通过leftOuterJoin操作既保留了左侧用户广告点击内容的RDD的所有内容,

// 又获得了相应点击内容是否在黑名单中

val joinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)

/**

* 进行filter过滤的时候,其输入元素是一个Tuple:(name,((time,name), boolean))

* 其中第一个元素是黑名单的名称,第二元素的第二个元素是进行leftOuterJoin的时候是否存在的值。

* 如果存在的话,表面当前广告点击是黑名单,需要过滤掉,否则的话是有效点击内容;

*/

val validClicked = joinedBlackListRDD.filter(joinedItem => {

if(joinedItem._2._2.getOrElse(false))

{

false

} else {

true

}

})

validClicked.map(validClick => {validClick._2._1})

}).print

/**

* 计算后的有效数据一般都会写入Kafka中,下游的计费系统会从kafka中pull到有效数据进行计费

*/

ssc.start()

ssc.awaitTermination()

}

}

我们把程序的Batch Interval设置成300秒:

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

我们重新生成一下jar包 。

Spark集群有4台机器:Master、Worker1、Worker2、Worker3、Worker4。

启动Spark的History Server。

打开数据发送的端口:

nc -lk 9999

用spark-submit运行前面生成的jar包。

在数据发送端口输入若干数据,比如:

1375864674543 Tom

1375864674553 Spy

1375864674571 Andy

1375864688436 Cheater

1375864784240 Kelvin

1375864853892 Steven

1375864979347 John

打开浏览器,看History Server的日志信息:

点击最新的应用,看我们目前运行的应用程序中有些什么Job:

总共竟然有5个Job。这完全不是我们此前做Spark SQL之类的应用程序时看到的样子。

我们接下来看一看这些Job的内容,主要揭示一些现象,不会做完全深入的剖析,只是为了先让大家进行一些思考。

Job 0:此Job不体现我们的业务逻辑代码。这个Job是出于对后面计算的负载均衡的考虑。

Job 0包含有Stage 0、Stage 1。随便看一个Stage,比如Stage 1。看看其中的Aggregated Metrics by Executor部分:

发现此Stage在所有Executor上都存在。

Job 1:运行时间比较长,耗时1.5分钟。

点击Stage 2的链接,进去看看Aggregated Metrics By Executor部分:

可以知道,Stage 2只在Worker4上的一个Executor执行,而且执行了1.5分钟。

是否会觉得奇怪:从业务处理的角度看,我们发送的那么一点数据,没有必要去启动一个运行1.5分钟的任务吧。那这个任务是做什么呢?

从DAG Visualization部分,就知道此Job实际就是启动了一个接收数据的Receiver:

原来Receiver是通过一个Job来启动的。那肯定有一个Action来触发它。

看看Tasks部分:

只有一个Worker运行此Job。是用于接收数据。

Locality Level是PROCESS_LOCAL,原来是内存节点。所以,默认情况下,数据接收不会使用磁盘,而是直接使用内存中的数据。

看来,Spark Streaming应用程序启动后,自己会启动一些Job。默认启动了一个Job来接收数据,为后续处理做准备。

重要启示:一个Spark应用程序中可以启动很多Job,而这些不同的Job之间可以相互配合。这一认识为我们写复杂Spark程序奠定了良好的基础。

Job 2:看Details可以发现有我们程序的主要业务逻辑,体现在Stag 3、Stag4、Stag 5中。

我们看Stag3、Stage4的详情,可以知道这2个Stage都是用4个Executor执行的。所有数据处理是在4台机器上进行的。

Stag 5只在Worker4上。这是因为这个Stage有Shuffle操作。

Job3:有Stage 6、Stage 7、Stage 8。其中Stage 6、Stage 7被跳过。

看看Stage 8的Aggregated Metrics by Executor部分。可以看到,数据处理是在4台机器上进行的:

Job4:也体现了我们应用程序中的业务逻辑 。有Stage 9、Stage 10、Stage 11。其中Stage 9、Stage 10被跳过。

  

看看Stage 11的详情。可以看到,数据处理是在Worker4之外的其它3台机器上进行的:

综合以上的现象可以知道,Spark Streaming的一个应用中,运行了这么多Job,远不是我们从网络博客或者书籍上看的那么简单。

我们有必要通过这些现象,反过来回溯去寻根问源。不过这次暂不做深入分析。

我们的神奇之旅才刚刚开始。

2 瞬间理解Spark Streaming本质

以上的连续4个图,分别对应以下4个段落的描述:

Spark Streaming接收Kafka、Flume、HDFS和Kinesis等各种来源的实时输入数据,进行处理后,处理结果保存在HDFS、Databases等各种地方。

Spark Streaming接收这些实时输入数据流,会将它们按批次划分,然后交给Spark引擎处理,生成按照批次划分的结果流。

Spark Streaming提供了表示连续数据流的、高度抽象的被称为离散流的DStream。DStream本质上表示RDD的序列。任何对DStream的操作都会转变为对底层RDD的操作。

Spark Streaming使用数据源产生的数据流创建DStream,也可以在已有的DStream上使用一些操作来创建新的DStream。

在我们前面的实验中,每300秒会产生一批数据,基于这批数据会生成RDD,进而触发Job,执行处理。

DStream是一个没有边界的集合,没有大小的限制。

DStream代表了时空的概念。随着时间的推移,里面不断产生RDD。

锁定到时间段后,就是空间的操作。也就是对本时间段的对应批次的数据的处理。

下面用实例来讲解数据处理过程。

数据处理会有若干个对DStream的操作,这些操作之间的依赖关系,构成了DStreamGraph。如以下图例所示:

上图中每个foreach都会触发一个作业,就会顺着依赖从后往前回溯,形成DAG,如下图所示:

空间维度确定之后,随着时间不断推进,会不断实例化RDD Graph,然后触发Job去执行处理。

现在再去读官方的Spark Streaming的文档,就好理解多了。

看来我们的学习,将从Spark Streaming的现象开始,深入到Spark Core和Spark Streaming的本质。

时间: 2024-10-08 21:32:25

Dream_Spark版本定制第一课的相关文章

Spark源码定制第一课:通过案例对SparkStreaming透彻理解三板斧之一

第一课:通过案例对SparkStreaming透彻理解三板斧之一:解密SparkStreaming另类实验及SparkStreaming本质解析 本期导读: 1 Spark源码定制选择从SparkStreaming入手: 2 Spark Streaming另类在线实验: 3 瞬间理解SparkStreaming本质. 1.    从Spark Streaming入手开始Spark源码版本定制之路 1.1           从Spark Streaming入手Spark源码版本定制之路的理由 从

(版本定制)第5课:基于案例分析Spark Streaming流计算框架的运行源码

本期内容: 1.在线动态计算分类最热门商品案例回顾与演示 2.基于案例分析Spark Streaming的运行源码 第一部分案例: package com.dt.spark.sparkstreaming import com.robinspark.utils.ConnectionPoolimport org.apache.spark.SparkConfimport org.apache.spark.sql.Rowimport org.apache.spark.sql.hive.HiveConte

(版本定制)第11课:Spark Streaming源码解读之Driver中的ReceiverTracker彻底研究和思考

本期内容: 1.ReceiverTracker的架构设计 2.消息循环系统 3.ReceiverTracker具体实现 上节课讲到了Receiver是如何不断的接收数据的,并且接收到的数据的元数据会汇报给ReceiverTracker,下面我们看看ReceiverTracker具体的功能及实现. ReceiverTracker主要的功能: 在Executor上启动Receivers. 停止Receivers . 更新Receiver接收数据的速度(也就是限流) 不断的等待Receivers的运行

(版本定制)第18课:Spark Streaming中空RDD处理及流处理程序优雅的停止

本期内容: 1. Spark Streaming中RDD为空处理 2. Streaming Context程序停止方式 Spark Streaming运用程序是根据我们设定的Batch Duration来产生RDD,产生的RDD存在partitons数据为空的情况,但是还是会执行foreachPartition,会获取计算资源,然后计算一下,这种情况就会浪费 集群计算资源,所以需要在程序运行的时候进行过滤,参考如下代码: package com.dt.spark.sparkstreamingim

版本定制第5课:基于案例一节课贯通Spark Streaming流计算框架的运行源码

本期内容: 1.在线动态计算分类最热门商品案例回顾与演示 2.基于案例贯通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.HiveCo

(版本定制)第14课:Spark Streaming源码解读之State管理之updateStateByKey和mapWithState解密

本期内容: 1.updateStateByKey解密 2.mapWithState解密 背景:整个Spark Streaming是按照Batch Duractions划分Job的.但是很多时候我们需要算过去的一天甚至一周的数据,这个时候不可避免的要进行状态管理,而Spark Streaming每个Batch Duractions都会产生一个Job,Job里面都是RDD, 所以此时面临的问题就是怎么对状态进行维护?这个时候就需要借助updateStateByKey和mapWithState方法完成

(版本定制)第16课:Spark Streaming源码解读之数据清理内幕彻底解密

本期内容: 1.Spark Streaming元数据清理详解 2.Spark Streaming元数据清理源码解析 一.如何研究Spark Streaming元数据清理 操作DStream的时候会产生元数据,所以要解决RDD的数据清理工作就一定要从DStream入手.因为DStream是RDD的模板,DStream之间有依赖关系. DStream的操作产生了RDD,接收数据也靠DStream,数据的输入,数据的计算,输出整个生命周期都是由DStream构建的.由此,DStream负责RDD的整个

(版本定制)第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析

本期内容: 1.Spark Streaming资源动态分配 2.Spark Streaming动态控制消费速率 为什么需要动态? a)Spark默认情况下粗粒度的,先分配好资源再计算.对于Spark Streaming而言有高峰值和低峰值,但是他们需要的资源是不一样的,如果按照高峰值的角度的话,就会有大量的资源浪费. b) Spark Streaming不断的运行,对资源消耗和管理也是我们要考虑的因素. Spark Streaming资源动态调整的时候会面临挑战: Spark Streaming

(版本定制)第15课:Spark Streaming源码解读之No Receivers彻底思考

hu本期内容: 1.Kafka解密 背景: 目前No Receivers在企业中使用的越来越多,No Receivers具有更强的控制度,语义一致性.No Receivers是我们操作数据来源自然方式,操作数据来源使用一个封装器,且是RDD类型的. 所以Spark Streaming就产生了自定义RDD –> KafkaRDD. 源码分析: 1.KafkaRDD源码 private[kafka]class KafkaRDD[K: ClassTag,V: ClassTag,U <: Decode