RDD(弹性分布式数据集，RDD里面并不存放真正的数据，对RDD进行操作会在Driver端转换成Task，下发到Executor计算分散在多台机器上的数据；

RDD是一个代理，对代理进行操作，他会生成Task,帮你计算；

操作代理就像操作本地集合一样；

RDD(Resilient Distributed Dataset)是Spark的最基本抽象，是对分布式内存的抽象使用，实现了以操作本地集合的方式来操作分布式数据集的抽象实现。

创建RDD，然后对RDD进行操作（调用rdd的方法，方法分为两类，一个叫transformation，一类叫Action）

创建RDD的方式：

1通过外部存储系统创建RDD

2调用一个已经存在的RDD的Transformation，会生成一个新的RDD；

RDD的transformnation（RDD的常用算子）：

parallelize mkrdd textfile

map()

distinct()

flattenmap()

union() 合并两个RDD

rdd6.intersection(rdd7) 求交集

reduceByKey()

groupBtKey()

#join(连接）

rdd6 join rdd7

rdd6.leftOuterJoin(rdd7)

rdd6.rightOuterJoin(rdd7)

rdd6.cogroup(rdd7)

first() 返回第一个元素

tark(n) 返回第n个元素

collectt() 返回 RDD 中的所有元素

count () 返回 RDD 中的元素个数

rdd.top(n) 按照降序的或者指定的排序规则，返回前n个元素

foreach() 对 RDD 中的每个元素使用给定的函数

Spark Word Count的详解

1.textFile()是用hadoopFile()函数读取数据，生成HadoopRDD,格式是TextInputFormat，和Hadoop map函数读取数据的函数一致（一行行的读取数据），然后调用map（）函数生成最终的mappartionRDD。textFile()生成两个RDD。

spark的cache() 把数据添加到内存，提高程序的效率。

val cache = data.cache() 当数据较大时，cache()只能cache部分数据到内存

Spark为持久化RDD定义了几种不同的机制，用不同的StorageLevel值表示。rdd.cache()是rdd.persist(StorageLevel.MEMORY)的简写，它将RDD存储为未序列化的Java对象。

cache只有一个默认的缓存级别MEMORY_ONLY ，而persist可以根据情况设置其它的缓存级别。

//缓存类型

NONE :什么类型都不是
DISK_ONLY：磁盘
DISK_ONLY_2：磁盘；双副本
MEMORY_ONLY： 内存；反序列化；把RDD作为反序列化的方式存储，假如RDD的内容存不下，剩余的分区在以后需要时会重新计算，不会刷到磁盘上。
MEMORY_ONLY_2：内存；反序列化；双副本
MEMORY_ONLY_SER：内存；序列化；这种序列化方式，每一个partition以字节数据存储，好处是能带来更好的空间存储，但CPU耗费高
MEMORY_ONLY_SER_2 : 内存；序列化；双副本
MEMORY_AND_DISK：内存 + 磁盘；反序列化；双副本；RDD以反序列化的方式存内存，假如RDD的内容存不下，剩余的会存到磁盘
MEMORY_AND_DISK_2 : 内存 + 磁盘；反序列化；双副本
MEMORY_AND_DISK_SER：内存 + 磁盘；序列化
MEMORY_AND_DISK_SER_2：内存 + 磁盘；序列化；双副本
*********** 序列化能有效减少存储空间，默认MEMORY_ONLY

spark的广播变量

Spark中因为算子中的真正逻辑是发送到Executor中去运行的，所以当Executor中需要引用外部变量时，需要使用广播变量。

• 注意事项

1、能不能将一个RDD使用广播变量广播出去？

不能，因为RDD是不存储数据的。可以将RDD的结果广播出去。

2、广播变量只能在Driver端定义，不能在Executor端定义。

3、在Driver端可以修改广播变量的值，在Executor端无法修改广播变量的值。

4、如果executor端用到了Driver的变量，如果不使用广播变量在Executor有多少task就有多少Driver端的变量副本。

5、如果Executor端用到了Driver的变量，如果使用广播变量在每个Executor中只有一份Driver端的变量副本。

广播变量的好处

广播变量的好处，不是每个task一份变量副本，而是变成每个节点的executor才一份副本。这样的话，
就可以让变量产生的副本大大减少。

Spark执行过程简介

1.构建Spark Application的运行环境（启动SparkContext），SparkContext向资源管理器（可以是Standalone、Mesos或YARN）注册并申请运行Executor资源；
2.资源管理器分配Executor资源并启动StandaloneExecutorBackend，Executor运行情况将随着心跳发送到资源管理器上；
3.SparkContext构建成DAG图，将DAG图分解成Stage，并把Taskset发送给Task Scheduler。Executor向SparkContext申请Task，Task Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor。
4.Task在Executor上运行，运行完毕释放所有资源。

spark-yarn

1.Spark Yarn Client向YARN中提交应用程序，包括ApplicationMaster程序、启动ApplicationMaster的命令、需要在Executor中运行的程序等；
2.ResourceManager收到请求后，在集群中选择一个NodeManager，为该应用程序分配第一个Container，要求它在这个 Container中启动应用程序的ApplicationMaster，其中ApplicationMaster进行SparkContext等的初始 化；
3.ApplicationMaster向ResourceManager注册，这样用户可以直接通过ResourceManage查看应用程序的运行状态，然后它将采用轮询的方式通过RPC协议为各个任务申请资源，并监控它们的运行状态直到运行结束；
4.一旦ApplicationMaster申请到资源（也就是Container）后，便与对应的NodeManager通信，要求它在获得的 Container中启动启动CoarseGrainedExecutorBackend，CoarseGrainedExecutorBackend启 动后会向ApplicationMaster中的SparkContext注册并申请Task。这一点和Standalone模式一样，只不过 SparkContext在Spark Application中初始化时，使用CoarseGrainedSchedulerBackend配合YarnClusterScheduler进行 任务的调度，其中YarnClusterScheduler只是对TaskSchedulerImpl的一个简单包装，增加了对Executor的等待逻 辑等；
5.ApplicationMaster中的SparkContext分配Task给CoarseGrainedExecutorBackend执 行，CoarseGrainedExecutorBackend运行Task并向ApplicationMaster汇报运行的状态和进度，以让 ApplicationMaster随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务；
6.应用程序运行完成后，ApplicationMaster向ResourceManager申请注销并关闭自己。

spark切分stage

客户端构建好RDD的DAG以后，会提交至DAGScheduler来处理，这是一个Stage级别的调度器，他首先会把作业切分为一个个Stage，每个Stage由一组相同运算的tasks组成，然后会以taskset的形式提交给TaskScheduler。DS还会跟踪stage的输出与物化情况、检测task运行时的最优位置，重新提交失败的stage。

窄依赖和宽依赖
窄依赖：
指父RDD的每一个分区最多被一个子RDD的分区所用，表现为一个父RDD的分区对应于一个子RDD的分区，和两个父RDD的分区对应于一个子RDD 的分区。图中，map/filter和union属于第一类，对输入进行协同划分（co-partitioned）的join属于第二类。

宽依赖：
指子RDD的分区依赖于父RDD的所有分区，这是因为shuffle类操作，如图中的groupByKey和未经协同划分的join。

Stage:
一个Job会被拆分为多组Task，每组任务被称为一个Stage就像Map Stage， Reduce Stage。Stage的划分在RDD的论文中有详细的介绍，简单的说是以shuffle和result这两种类型来划分。在Spark中有两类task，一类是shuffleMapTask，一类是resultTask，第一类task的输出是shuffle所需数据，第二类task的输出是result，stage的划分也以此为依据，shuffle之前的所有变换是一个stage，shuffle之后的操作是另一个stage。比如 rdd.parallize(1 to 10).foreach(println) 这个操作没有shuffle，直接就输出了，那么只有它的task是resultTask，stage也只有一个；如果是rdd.map(x => (x, 1)).reduceByKey(_ + _).foreach(println), 这个job因为有reduce，所以有一个shuffle过程，那么reduceByKey之前的是一个stage，执行shuffleMapTask，输出shuffle所需的数据，reduceByKey到最后是一个stage，直接就输出结果了。如果job中有多次shuffle，那么每个shuffle之前都是一个stage.
会根据RDD之间的依赖关系将DAG图划分为不同的阶段，对于窄依赖，由于partition依赖关系的确定性，partition的转换处理就可以在同一个线程里完成，窄依赖就被spark划分到同一个stage中，而对于宽依赖，只能等父RDD shuffle处理完成后，下一个stage才能开始接下来的计算。之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中
Stage划分思路
因此spark划分stage的整体思路是：从后往前推，遇到宽依赖就断开，划分为一个stage；遇到窄依赖就将这个RDD加入该stage中。因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。
　　在spark中，Task的类型分为2种：ShuffleMapTask和ResultTask；简单来说，DAG的最后一个阶段会为每个结果的partition生成一个ResultTask，即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的！而其余所有阶段都会生成ShuffleMapTask；之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中；也就是说图2中的stage1和stage2相当于mapreduce中的Mapper,而ResultTask所代表的stage3就相当于mapreduce中的reducer。

sparkStreaming的原理

对于Spark Core它的核心就是RDD，对于Spark Streaming来说，它的核心是DStream，DStream类似于RDD，它实质上一系列的RDD的集合，DStream可以按照秒数将数据流进行批量的划分。首先从接收到流数据之后，将其划分为多个batch，然后提交给Spark集群进行计算，最后将结果批量输出到HDFS或者数据库以及前端页面展示等等。对于DStream如何理解呢？它是一系列连续的RDD，它是建立在Spark之上的不可变的，分布式数据集，在DStream中的每一个RDD包含着一定时间间隔的数据，如下图所示：

那么，Spark Streaming的工作原理是什么呢？它是怎么运行在集群上的呢？其原理架构图如下所示：

我们都知道Spark Core在初始化时会生成一个SparkContext对象来对数据进行后续的处理，相对应的Spark Streaming会创建一个Streaming Context，它的底层是SparkContext，也就是说它会将任务提交给SparkContext来执行，这也很好的解释了DStream是一系列的RDD。当启动Spark Streaming应用的时候，首先会在一个节点的Executor上启动一个Receiver接受者，然后当从数据源写入数据的时候会被Receiver接收，接收到数据之后Receiver会将数据Split成很多个block，然后备份到各个节点（Replicate Blocks 容灾恢复），然后Receiver向StreamingContext进行块报告，说明数据在那几个节点的Executor上，接着在一定间隔时间内StreamingContext会将数据处理为RDD并且交给SparkContext划分到各个节点进行并行计算。

Spark的各个子框架，都是基于核心Spark的，Spark Streaming在内部的处理机制是，接收实时流的数据，并根据一定的时间间隔拆分成一批批的数据，然后通过Spark Engine处理这些批数据，最终得到处理后的一批批结果数据。

对应的批数据，在Spark内核对应一个RDD实例，因此，对应流数据的DStream可以看成是一组RDDs，即RDD的一个序列。通俗点理解的话，在流数据分成一批一批后，通过一个先进先出的队列，然后Spark Engine从该队列中依次取出一个个批数据，把批数据封装成一个RDD，然后进行处理，这是一个典型的生产者消费者模型，对应的就有生产者消费者模型的问题，即如何协调生产速率和消费速率。

1.2术语定义

l离散流（discretized stream）或DStream：这是Spark Streaming对内部持续的实时数据流的抽象描述，即我们处理的一个实时数据流，在Spark Streaming中对应于一个DStream实例。

l批数据（batch data）：这是化整为零的第一步，将实时流数据以时间片为单位进行分批，将流处理转化为时间片数据的批处理。随着持续时间的推移，这些处理结果就形成了对应的结果数据流了。

l时间片或批处理时间间隔（batch interval）：这是人为地对流数据进行定量的标准，以时间片作为我们拆分流数据的依据。一个时间片的数据对应一个RDD实例。

l窗口长度（window length）：一个窗口覆盖的流数据的时间长度。必须是批处理时间间隔的倍数，

l滑动时间间隔：前一个窗口到后一个窗口所经过的时间长度。必须是批处理时间间隔的倍数

lInput DStream:一个input DStream是一个特殊的DStream，将Spark Streaming连接到一个外部数据源来读取数据。

2、运行原理

2.1Streaming架构

SparkStreaming是一个对实时数据流进行高通量、容错处理的流式处理系统，可以对多种数据源（如Kdfka、Flume、Twitter、Zero和TCP套接字）进行类似Map、Reduce和Join等复杂操作，并将结果保存到外部文件系统、数据库或应用到实时仪表盘。

l计算流程：Spark Streaming是将流式计算分解成一系列短小的批处理作业。这里的批处理引擎是Spark Core，也就是把Spark Streaming的输入数据按照batch size（如1秒）分成一段一段的数据（Discretized Stream），每一段数据都转换成Spark中的RDD（Resilient Distributed Dataset），然后将Spark Streaming中对DStream的Transformation操作变为针对Spark中对RDD的Transformation操作，将RDD经过操作变成中间结果保存在内存中。整个流式计算根据业务的需求可以对中间的结果进行叠加或者存储到外部设备。下图显示了Spark Streaming的整个流程。

图Spark Streaming构架

2.2.1如何使用Spark Streaming

1.创建StreamingContext对象同Spark初始化需要创建SparkContext对象一样，使用Spark Streaming就需要创建StreamingContext对象。创建StreamingContext对象所需的参数与SparkContext基本一致，包括指明Master，设定名称(如NetworkWordCount)。需要注意的是参数Seconds(1)，Spark Streaming需要指定处理数据的时间间隔，如上例所示的1s，那么Spark Streaming会以1s为时间窗口进行数据处理。此参数需要根据用户的需求和集群的处理能力进行适当的设置；

2.创建InputDStream如同Storm的Spout，Spark Streaming需要指明数据源。如上例所示的socketTextStream，Spark Streaming以socket连接作为数据源读取数据。当然Spark Streaming支持多种不同的数据源，包括Kafka、Flume、HDFS/S3、Kinesis和Twitter等数据源；

3.操作DStream对于从数据源得到的DStream，用户可以在其基础上进行各种操作，如上例所示的操作就是一个典型的WordCount执行流程：对于当前时间窗口内从数据源得到的数据首先进行分割，然后利用Map和ReduceByKey方法进行计算，当然最后还有使用print()方法输出结果；

4.启动Spark Streaming之前所作的所有步骤只是创建了执行流程，程序没有真正连接上数据源，也没有对数据进行任何操作，只是设定好了所有的执行计划，当ssc.start()启动后程序才真正进行所有预期的操作。

至此对于Spark Streaming的如何使用有了一个大概的印象，在后面的章节我们会通过源代码深入探究一下Spark Streaming的执行流程。

2.2.3DStream的操作

与RDD类似，DStream也提供了自己的一系列操作方法，这些操作可以分成三类：普通的转换操作、窗口转换操作和输出操作。

2.2.3.1普通的转换操作

普通的转换操作如下表所示：

转换	描述
map(func)	源DStream的每个元素通过函数func返回一个新的DStream。
flatMap(func)	类似与map操作，不同的是每个输入元素可以被映射出0或者更多的输出元素。
filter(func)	在源DSTREAM上选择Func函数返回仅为true的元素,最终返回一个新的DSTREAM。
repartition(numPartitions)	通过输入的参数numPartitions的值来改变DStream的分区大小。
union(otherStream)	返回一个包含源DStream与其他DStream的元素合并后的新DSTREAM。
count()	对源DStream内部的所含有的RDD的元素数量进行计数，返回一个内部的RDD只包含一个元素的DStreaam。
reduce(func)	使用函数func（有两个参数并返回一个结果）将源DStream中每个RDD的元素进行聚 合操作,返回一个内部所包含的RDD只有一个元素的新DStream。
countByValue()	计算DStream中每个RDD内的元素出现的频次并返回新的DStream[(K,Long)]，其中K是RDD中元素的类型，Long是元素出现的频次。
reduceByKey(func, [numTasks])	当一个类型为（K，V）键值对的DStream被调用的时候,返回类型为类型为（K，V）键值对的新DStream,其中每个键的值V都是使用聚合函数func汇总。注意：默认情况下，使用Spark的默认并行度提交任务（本地模式下并行度为2，集群模式下位8），可以通过配置numTasks设置不同的并行任务数。
join(otherStream, [numTasks])	当被调用类型分别为（K，V）和（K，W）键值对的2个DStream时，返回类型为（K，（V，W））键值对的一个新DSTREAM。
cogroup(otherStream, [numTasks])	当被调用的两个DStream分别含有(K, V)和(K, W)键值对时,返回一个(K, Seq[V], Seq[W])类型的新的DStream。
transform(func)	通过对源DStream的每RDD应用RDD-to-RDD函数返回一个新的DStream，这可以用来在DStream做任意RDD操作。
updateStateByKey(func)	返回一个新状态的DStream,其中每个键的状态是根据键的前一个状态和键的新值应用给定函数func后的更新。这个方法可以被用来维持每个键的任何状态数据。

在上面列出的这些操作中，transform()方法和updateStateByKey()方法值得我们深入的探讨一下：

ltransform(func)操作

该transform操作（转换操作）连同其其类似的transformWith操作允许DStream上应用任意RDD-to-RDD函数。它可以被应用于未在DStream API中暴露任何的RDD操作。例如，在每批次的数据流与另一数据集的连接功能不直接暴露在DStream API中，但可以轻松地使用transform操作来做到这一点，这使得DStream的功能非常强大。例如，你可以通过连接预先计算的垃圾邮件信息的输入数据流（可能也有Spark生成的），然后基于此做实时数据清理的筛选，如下面官方提供的伪代码所示。事实上，也可以在transform方法中使用机器学习和图形计算的算法。

lupdateStateByKey操作

该updateStateByKey操作可以让你保持任意状态，同时不断有新的信息进行更新。要使用此功能，必须进行两个步骤 ：

（1）定义状态-状态可以是任意的数据类型。

（2）定义状态更新函数-用一个函数指定如何使用先前的状态和从输入流中获取的新值 更新状态。

让我们用一个例子来说明，假设你要进行文本数据流中单词计数。在这里，正在运行的计数是状态而且它是一个整数。我们定义了更新功能如下：

此函数应用于含有键值对的DStream中（如前面的示例中，在DStream中含有（word，1）键值对）。它会针对里面的每个元素（如wordCount中的word）调用一下更新函数，newValues是最新的值，runningCount是之前的值。

2.2.3.2窗口转换操作

Spark Streaming还提供了窗口的计算，它允许你通过滑动窗口对数据进行转换，窗口转换操作如下：

转换	描述
window(windowLength,slideInterval)	返回一个基于源DStream的窗口批次计算后得到新的DStream。
countByWindow(windowLength,slideInterval)	返回基于滑动窗口的DStream中的元素的数量。
reduceByWindow(func,windowLength,slideInterval)	基于滑动窗口对源DStream中的元素进行聚合操作，得到一个新的DStream。
reduceByKeyAndWindow(func,windowLength,slideInterval, [numTasks])	基于滑动窗口对（K，V）键值对类型的DStream中的值按K使用聚合函数func进行聚合操作，得到一个新的DStream。
reduceByKeyAndWindow(func,invFunc,windowLength,slideInterval, [numTasks])	一个更高效的reduceByKkeyAndWindow()的实现版本，先对滑动窗口中新的时间间隔内数据增量聚合并移去最早的与新增数据量的时间间隔内的数据统计量。例如，计算t+4秒这个时刻过去5秒窗口的WordCount，那么我们可以将t+3时刻过去5秒的统计量加上[t+3，t+4]的统计量，在减去[t-2，t-1]的统计量，这种方法可以复用中间三秒的统计量，提高统计的效率。
countByValueAndWindow(windowLength,slideInterval, [numTasks])	基于滑动窗口计算源DStream中每个RDD内每个元素出现的频次并返回DStream[(K,Long)]，其中K是RDD中元素的类型，Long是元素频次。与countByValue一样，reduce任务的数量可以通过一个可选参数进行配置。

批处理间隔示意图

在Spark Streaming中，数据处理是按批进行的，而数据采集是逐条进行的，因此在Spark Streaming中会先设置好批处理间隔（batch duration），当超过批处理间隔的时候就会把采集到的数据汇总起来成为一批数据交给系统去处理。

对于窗口操作而言，在其窗口内部会有N个批处理数据，批处理数据的大小由窗口间隔（window duration）决定，而窗口间隔指的就是窗口的持续时间，在窗口操作中，只有窗口的长度满足了才会触发批数据的处理。除了窗口的长度，窗口操作还有另一个重要的参数就是滑动间隔（slide duration），它指的是经过多长时间窗口滑动一次形成新的窗口，滑动窗口默认情况下和批次间隔的相同，而窗口间隔一般设置的要比它们两个大。在这里必须注意的一点是滑动间隔和窗口间隔的大小一定得设置为批处理间隔的整数倍。

如批处理间隔示意图所示，批处理间隔是1个时间单位，窗口间隔是3个时间单位，滑动间隔是2个时间单位。对于初始的窗口time 1-time 3，只有窗口间隔满足了才触发数据的处理。这里需要注意的一点是，初始的窗口有可能流入的数据没有撑满，但是随着时间的推进，窗口最终会被撑满。当每个2个时间单位，窗口滑动一次后，会有新的数据流入窗口，这时窗口会移去最早的两个时间单位的数据，而与最新的两个时间单位的数据进行汇总形成新的窗口（time3-time5）。

对于窗口操作，批处理间隔、窗口间隔和滑动间隔是非常重要的三个时间概念，是理解窗口操作的关键所在。

2.2.3.3输出操作

Spark Streaming允许DStream的数据被输出到外部系统，如数据库或文件系统。由于输出操作实际上使transformation操作后的数据可以通过外部系统被使用，同时输出操作触发所有DStream的transformation操作的实际执行（类似于RDD操作）。以下表列出了目前主要的输出操作：

转换	描述
print()	在Driver中打印出DStream中数据的前10个元素。
saveAsTextFiles(prefix, [suffix])	将DStream中的内容以文本的形式保存为文本文件，其中每次批处理间隔内产生的文件以prefix-TIME_IN_MS[.suffix]的方式命名。
saveAsObjectFiles(prefix, [suffix])	将DStream中的内容按对象序列化并且以SequenceFile的格式保存。其中每次批处理间隔内产生的文件以prefix-TIME_IN_MS[.suffix]的方式命名。
saveAsHadoopFiles(prefix, [suffix])	将DStream中的内容以文本的形式保存为Hadoop文件，其中每次批处理间隔内产生的文件以prefix-TIME_IN_MS[.suffix]的方式命名。
foreachRDD(func)	最基本的输出操作，将func函数应用于DStream中的RDD上，这个操作会输出数据到外部系统，比如保存RDD到文件或者网络数据库等。需要注意的是func函数是在运行该streaming应用的Driver进程里执行的。

dstream.foreachRDD是一个非常强大的输出操作，它允将许数据输出到外部系统。但是 ，如何正确高效地使用这个操作是很重要的，下面展示了如何去避免一些常见的错误。

通常将数据写入到外部系统需要创建一个连接对象（如TCP连接到远程服务器），并用它来发送数据到远程系统。出于这个目的，开发者可能在不经意间在Spark driver端创建了连接对象，并尝试使用它保存RDD中的记录到Spark worker上，如下面代码：

这是不正确的，这需要连接对象进行序列化并从Driver端发送到Worker上。连接对象很少在不同机器间进行这种操作，此错误可能表现为序列化错误（连接对不可序列化），初始化错误（连接对象在需要在Worker上进行需要初始化） 等等，正确的解决办法是在worker上创建的连接对象。

通常情况下，创建一个连接对象有时间和资源开销。因此，创建和销毁的每条记录的连接对象可能招致不必要的资源开销，并显著降低系统整体的吞吐量 。一个更好的解决方案是使用rdd.foreachPartition方法创建一个单独的连接对象，然后使用该连接对象输出的所有RDD分区中的数据到外部系统。

这缓解了创建多条记录连接的开销。最后，还可以进一步通过在多个RDDs/ batches上重用连接对象进行优化。一个保持连接对象的静态池可以重用在多个批处理的RDD上将其输出到外部系统，从而进一步降低了开销。

需要注意的是，在静态池中的连接应该按需延迟创建，这样可以更有效地把数据发送到外部系统。另外需要要注意的是：DStreams延迟执行的，就像RDD的操作是由actions触发一样。默认情况下，输出操作会按照它们在Streaming应用程序中定义的顺序一个个执行。

2.3容错、持久化和性能调优

2.3.1容错

DStream基于RDD组成，RDD的容错性依旧有效，我们首先回忆一下SparkRDD的基本特性。

lRDD是一个不可变的、确定性的可重复计算的分布式数据集。RDD的某些partition丢失了，可以通过血统（lineage）信息重新计算恢复；

l如果RDD任何分区因worker节点故障而丢失，那么这个分区可以从原来依赖的容错数据集中恢复；

l由于Spark中所有的数据的转换操作都是基于RDD的，即使集群出现故障，只要输入数据集存在，所有的中间结果都是可以被计算的。

Spark Streaming是可以从HDFS和S3这样的文件系统读取数据的，这种情况下所有的数据都可以被重新计算，不用担心数据的丢失。但是在大多数情况下，Spark Streaming是基于网络来接受数据的，此时为了实现相同的容错处理，在接受网络的数据时会在集群的多个Worker节点间进行数据的复制（默认的复制数是2），这导致产生在出现故障时被处理的两种类型的数据：

1）Data received and replicated：一旦一个Worker节点失效，系统会从另一份还存在的数据中重新计算。

2）Data received but buffered for replication：一旦数据丢失，可以通过RDD之间的依赖关系，从HDFS这样的外部文件系统读取数据。

此外，有两种故障，我们应该关心：

（1）Worker节点失效：通过上面的讲解我们知道，这时系统会根据出现故障的数据的类型，选择是从另一个有复制过数据的工作节点上重新计算，还是直接从从外部文件系统读取数据。

（2）Driver（驱动节点）失效 ：如果运行Spark Streaming应用时驱动节点出现故障，那么很明显的StreamingContext已经丢失，同时在内存中的数据全部丢失。对于这种情况，Spark Streaming应用程序在计算上有一个内在的结构——在每段micro-batch数据周期性地执行同样的Spark计算。这种结构允许把应用的状态（亦称checkpoint）周期性地保存到可靠的存储空间中，并在driver重新启动时恢复该状态。具体做法是在ssc.checkpoint(<checkpoint directory>)函数中进行设置，Spark Streaming就会定期把DStream的元信息写入到HDFS中，一旦驱动节点失效，丢失的StreamingContext会通过已经保存的检查点信息进行恢复。

最后我们谈一下Spark Stream的容错在Spark 1.2版本的一些改进：

实时流处理系统必须要能在24/7时间内工作，因此它需要具备从各种系统故障中恢复过来的能力。最开始，SparkStreaming就支持从driver和worker故障恢复的能力。然而有些数据源的输入可能在故障恢复以后丢失数据。在Spark1.2版本中，Spark已经在SparkStreaming中对预写日志（也被称为journaling）作了初步支持，改进了恢复机制，并使更多数据源的零数据丢失有了可靠。

对于文件这样的源数据，driver恢复机制足以做到零数据丢失，因为所有的数据都保存在了像HDFS或S3这样的容错文件系统中了。但对于像Kafka和Flume等其它数据源，有些接收到的数据还只缓存在内存中，尚未被处理，它们就有可能会丢失。这是由于Spark应用的分布操作方式引起的。当driver进程失败时，所有在standalone/yarn/mesos集群运行的executor，连同它们在内存中的所有数据，也同时被终止。对于Spark Streaming来说，从诸如Kafka和Flume的数据源接收到的所有数据，在它们处理完成之前，一直都缓存在executor的内存中。纵然driver重新启动，这些缓存的数据也不能被恢复。为了避免这种数据损失，在Spark1.2发布版本中引进了预写日志（WriteAheadLogs）功能。

预写日志功能的流程是：1）一个SparkStreaming应用开始时（也就是driver开始时），相关的StreamingContext使用SparkContext启动接收器成为长驻运行任务。这些接收器接收并保存流数据到Spark内存中以供处理。2）接收器通知driver。3）接收块中的元数据（metadata）被发送到driver的StreamingContext。这个元数据包括：（a）定位其在executor内存中数据的块referenceid，（b）块数据在日志中的偏移信息（如果启用了）。

用户传送数据的生命周期如下图所示。

类似Kafka这样的系统可以通过复制数据保持可靠性。允许预写日志两次高效地复制同样的数据：一次由Kafka，而另一次由SparkStreaming。Spark未来版本将包含Kafka容错机制的原生支持，从而避免第二个日志。

2.3.2持久化

与RDD一样，DStream同样也能通过persist()方法将数据流存放在内存中，默认的持久化方式是MEMORY_ONLY_SER，也就是在内存中存放数据同时序列化的方式，这样做的好处是遇到需要多次迭代计算的程序时，速度优势十分的明显。而对于一些基于窗口的操作，如reduceByWindow、reduceByKeyAndWindow，以及基于状态的操作，如updateStateBykey，其默认的持久化策略就是保存在内存中。

对于来自网络的数据源（Kafka、Flume、sockets等），默认的持久化策略是将数据保存在两台机器上，这也是为了容错性而设计的。

另外，对于窗口和有状态的操作必须checkpoint，通过StreamingContext的checkpoint来指定目录，通过Dtream的checkpoint指定间隔时间，间隔必须是滑动间隔（slide interval）的倍数。

2.3.3性能调优

1.优化运行时间

l增加并行度确保使用整个集群的资源，而不是把任务集中在几个特定的节点上。对于包含shuffle的操作，增加其并行度以确保更为充分地使用集群资源；

l减少数据序列化，反序列化的负担Spark Streaming默认将接受到的数据序列化后存储，以减少内存的使用。但是序列化和反序列话需要更多的CPU时间，因此更加高效的序列化方式（Kryo）和自定义的系列化接口可以更高效地使用CPU；

l设置合理的batch duration（批处理时间间）在Spark Streaming中，Job之间有可能存在依赖关系，后面的Job必须确保前面的作业执行结束后才能提交。若前面的Job执行的时间超出了批处理时间间隔，那么后面的Job就无法按时提交，这样就会进一步拖延接下来的Job，造成后续Job的阻塞。因此设置一个合理的批处理间隔以确保作业能够在这个批处理间隔内结束时必须的；

l减少因任务提交和分发所带来的负担通常情况下，Akka框架能够高效地确保任务及时分发，但是当批处理间隔非常小（500ms）时，提交和分发任务的延迟就变得不可接受了。使用Standalone和Coarse-grained Mesos模式通常会比使用Fine-grained Mesos模式有更小的延迟。

2.优化内存使用

l控制batch size（批处理间隔内的数据量）Spark Streaming会把批处理间隔内接收到的所有数据存放在Spark内部的可用内存区域中，因此必须确保当前节点Spark的可用内存中少能容纳这个批处理时间间隔内的所有数据，否则必须增加新的资源以提高集群的处理能力；

l及时清理不再使用的数据前面讲到Spark Streaming会将接受的数据全部存储到内部可用内存区域中，因此对于处理过的不再需要的数据应及时清理，以确保Spark Streaming有富余的可用内存空间。通过设置合理的spark.cleaner.ttl时长来及时清理超时的无用数据，这个参数需要小心设置以免后续操作中所需要的数据被超时错误处理；

l观察及适当调整GC策略GC会影响Job的正常运行，可能延长Job的执行时间，引起一系列不可预料的问题。观察GC的运行情况，采用不同的GC策略以进一步减小内存回收对Job运行的影响。