Spark Streaming支持实时数据流的可扩展(Scalable)、高吞吐(high-throughput)、容错(fault-tolerant)的流处理(stream processing)。

特性如下：

• 可线性伸缩至超过数百个节点;

• 实现亚秒级延迟处理;

• 可与 Spark 批处理和交互式处理无缝集成;

• 提供简单的API实现复杂算法;

• 更多的流方式支持，包括 Kafka、Flume、Kinesis、Twitter、ZeroMQ 等。

---

001、原理

Spark 在接收到实时输入数据流后，将数据划分成批次(divides the data into batches)，然后转给 Spark Engine 处理，按批次生成最后的结果流(generate the final stream of results in batches)。

---

002、API

DStream：

DStream(Discretized Stream，离散流)是 Spark Stream 提供的高级抽象连续数据流。

组成：一个 DStream 可看作一个 RDDs 序列。

核心思想：将计算作为一系列较小时间间隔的、状态无关的、确定批次的任务，每个时间间隔内接收的输入数据被可靠存储在集群中，作为一个输入数据集。

特性：一个高层次的函数式编程 API、强一致性以及高校的故障恢复。

应用程序模板：

模板1

模板2

Input DStream：

Input DStream 是一种从流式数据源获取原始数据流的 DStream，分为基本输入源(文件系统、Socket、Akka Actor、自定义数据源)和高级输入源(Kafka、Flume等)。

Receiver：

每个 Input DStream(文件流除外)都会对应一个单一的 Receiver对象，负责从数据源接收数据并存入 Spark 内存进行处理。应用程序中可创建多个 Input DStream 并行接收多个数据流。

每个 Receiver 是一个长期运行在Worker或者 Executor 上的 Task，所以会占用该应用程序的一个核(core)。如果分配给 Spark Streaming 应用程序的核数小于或等于 Input DStream 个数(即Receiver个数)，则只能接收数据，却没有能力全部处理(文件流除外，因为无需Receiver)。

Spark Streaming 已封装各种数据源，需要时参考官方文档。

Transformation Operation

常用Transformation

updateStateByKey(func)

updateStateByKey可对DStream中的数据按key做reduce，然后对各批次数据累加

WordCount的updateStateByKey版本

transform(func)

通过对原 DStream 的每个 RDD 应用转换函数，创建一个新的 DStream。

官方文档代码举例

Window operations

窗口操作：基于 window 对数据 transformation(个人认为与Storm的tick相似，但功能更强大)。

参数：窗口长度(window length)和滑动时间间隔(slide interval)必须是源DStream 批次间隔的倍数。

举例说明：窗口长度为3，滑动时间间隔为2;上一行是原始 DStream，下一行是窗口化的 DStream。

常见 window operation

官方文档代码举例

join(otherStream, [numTasks])

连接数据流

官方文档代码举例1

官方文档代码举例2

Output Operation

缓存与持久化：

通过 persist()将 DStream 中每个 RDD 存储在内存。

Window operations 会自动持久化在内存，无需显示调用 persist()。

通过网络接收的数据流(如Kafka、Flume、Socket、ZeroMQ、RocketMQ等)执行 persist()时，默认在两个节点上持久化序列化后的数据，实现容错。

Checkpoint：

用途：Spark 基于容错存储系统(如HDFS、S3)进行故障恢复。

分类：

元数据检查点：保存流式计算信息用于 Driver 运行节点的故障恢复，包括创建应用程序的配置、应用程序定义的 DStream operations、已入队但未完成的批次。

数据检查点：保存生成的 RDD。由于 stateful transformation 需要合并多个批次的数据，即生成的 RDD 依赖于前几个批次 RDD 的数据(dependency chain)，为缩短 dependency chain 从而减少故障恢复时间，需将中间 RDD 定期保存至可靠存储(如HDFS)。

使用时机：

Stateful transformation：updateStateByKey()以及 window operations。

需要 Driver 故障恢复的应用程序。

---

003、使用方法

Stateful transformation

需要 Driver 故障恢复的应用程序(以WordCount举例)：如果 checkpoint 目录存在，则根据 checkpoint 数据创建新 StreamingContext;否则(如首次运行)新建 StreamingContext。

checkpoint 时间间隔

方法：

原则：一般设置为滑动时间间隔的5-10倍。

分析：checkpoint 会增加存储开销、增加批次处理时间。当批次间隔较小(如1秒)时，checkpoint 可能会减小 operation 吞吐量;反之，checkpoint 时间间隔较大会导致 lineage 和 task 数量增长。

---

004、性能调优

降低批次处理时间：

数据接收并行度

增加 DStream：接收网络数据(如Kafka、Flume、Socket等)时会对数据反序列化再存储在 Spark，由于一个 DStream 只有 Receiver 对象，如果成为瓶颈可考虑增加 DStream。

设置“spark.streaming.blockInterval”参数：接收的数据被存储在 Spark 内存前，会被合并成 block，而 block 数量决定了Task数量;举例，当批次时间间隔为2秒且 block 时间间隔为200毫秒时，Task 数量约为10;如果Task数量过低，则浪费了 CPU 资源;推荐的最小block时间间隔为50毫秒。

显式对 Input DStream 重新分区：在进行更深层次处理前，先对输入数据重新分区。

数据处理并行度：reduceByKey、reduceByKeyAndWindow 等 operation 可通过设置“spark.default.parallelism”参数或显式设置并行度方法参数控制。

数据序列化：可配置更高效的 Kryo 序列化。

设置合理批次时间间隔

原则：处理数据的速度应大于或等于数据输入的速度，即批次处理时间大于或等于批次时间间隔。

方法：

先设置批次时间间隔为5-10秒以降低数据输入速度;

再通过查看 log4j 日志中的“Total delay”，逐步调整批次时间间隔，保证“Total delay”小于批次时间间隔。

内存调优

持久化级别：开启压缩，设置参数“spark.rdd.compress”。

GC策略：在Driver和Executor上开启CMS。