【注】该系列文章以及使用到安装包/测试数据 可以在《倾情大奉送–Spark入门实战系列》获取
1 Spark Streaming简介
1.1 概述
Spark Streaming 是Spark核心API的一个扩展，可以实现高吞吐量的、具备容错机制的实时流数据的处理。支持从多种数据源获取数据，包括Kafk、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets，从数据源获取数据之后，可以使用诸如map、reduce、join和window等高级函数进行复杂算法的处理。最后还可以将处理结果存储到文件系统，数据库和现场仪表盘。在“One Stack rule them all”的基础上，还可以使用Spark的其他子框架，如集群学习、图计算等，对流数据进行处理。
Spark Streaming处理的数据流图：
Spark的各个子框架，都是基于核心Spark的，Spark Streaming在内部的处理机制是，接收实时流的数据，并根据一定的时间间隔拆分成一批批的数据，然后通过Spark Engine处理这些批数据，最终得到处理后的一批批结果数据。
对应的批数据，在Spark内核对应一个RDD实例，因此，对应流数据的DStream可以看成是一组RDDs，即RDD的一个序列。通俗点理解的话，在流数据分成一批一批后，通过一个先进先出的队列，然后 Spark Engine从该队列中依次取出一个个批数据，把批数据封装成一个RDD，然后进行处理，这是一个典型的生产者消费者模型，对应的就有生产者消费者模型的问题，即如何协调生产速率和消费速率。
1.2 术语定义
1.3 Storm与Spark Streming比较
虽然说两者都可以在各自的集群框架中运行，但是Storm可以在Mesos上运行, 而Spark Streaming可以在YARN和Mesos上运行。
2 运行原理
2.1 Streaming架构
SparkStreaming是一个对实时数据流进行高通量、容错处理的流式处理系统，可以对多种数据源（如Kdfka、Flume、Twitter、Zero和TCP 套接字）进行类似Map、Reduce和Join等复杂操作，并将结果保存到外部文件系统、数据库或应用到实时仪表盘。
计算流程：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中RDD的容错机制。每一个RDD都是一个不可变的分布式可重算的数据集，其记录着确定性的操作继承关系（lineage），所以只要输入数据是可容错的，那么任意一个RDD的分区（Partition）出错或不可用，都是可以利用原始输入数据通过转换操作而重新算出的。
对于Spark Streaming来说，其RDD的传承关系如下图所示，图中的每一个椭圆形表示一个RDD，椭圆形中的每个圆形代表一个RDD中的一个Partition，图中的每一列的多个RDD表示一个DStream（图中有三个DStream），而每一行最后一个RDD则表示每一个Batch Size所产生的中间结果RDD。我们可以看到图中的每一个RDD都是通过lineage相连接的，由于Spark Streaming输入数据可以来自于磁盘，例如HDFS（多份拷贝）或是来自于网络的数据流（Spark Streaming会将网络输入数据的每一个数据流拷贝两份到其他的机器）都能保证容错性，所以RDD中任意的Partition出错，都可以并行地在其他机器上将缺失的Partition计算出来。这个容错恢复方式比连续计算模型（如Storm）的效率更高。
实时性：对于实时性的讨论，会牵涉到流式处理框架的应用场景。Spark Streaming将流式计算分解成多个Spark Job，对于每一段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程。对于目前版本的Spark Streaming而言，其最小的Batch Size的选取在0.5~2秒钟之间（Storm目前最小的延迟是100ms左右），所以Spark Streaming能够满足除对实时性要求非常高（如高频实时交易）之外的所有流式准实时计算场景。
2.2 编程模型
DStream（Discretized Stream）作为Spark Streaming的基础抽象，它代表持续性的数据流。这些数据流既可以通过外部输入源赖获取，也可以通过现有的Dstream的transformation操作来获得。在内部实现上，DStream由一组时间序列上连续的RDD来表示。每个RDD都包含了自己特定时间间隔内的数据流。如下图所示。
对DStream中数据的各种操作也是映射到内部的RDD上来进行的，如下图所示，对Dtream的操作可以通过RDD的transformation生成新的DStream。这里的执行引擎是Spark。
2.2.1 如何使用Spark Streaming
作为构建于Spark之上的应用框架，Spark Streaming承袭了Spark的编程风格，对于已经了解Spark的用户来说能够快速地上手。接下来以Spark Streaming官方提供的WordCount代码为例来介绍Spark Streaming的使用方式。
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent from a starvation scenario.
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))
// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)
// Split each line into words
val words = lines.flatMap(_.split(" "))
import org.apache.spark.streaming.StreamingContext._
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)
// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate2.2.2 DStream的输入源
在Spark Streaming中所有的操作都是基于流的，而输入源是这一系列操作的起点。输入 DStreams 和 DStreams 接收的流都代表输入数据流的来源，在Spark Streaming 提供两种内置数据流来源：
2.2.2.1 基础来源
在前面分析怎样使用Spark Streaming的例子中我们已看到ssc.socketTextStream()方法，可以通过 TCP 套接字连接，从从文本数据中创建了一个 DStream。除了套接字，StreamingContext 的API还提供了方法从文件和 Akka actors 中创建 DStreams作为输入源。
Spark Streaming提供了streamingContext.fileStream(dataDirectory)方法可以从任何文件系统(如：HDFS、S3、NFS 等）的文件中读取数据，然后创建一个DStream。Spark Streaming 监控 dataDirectory 目录和在该目录下任何文件被创建处理(不支持在嵌套目录下写文件)。需要注意的是：读取的必须是具有相同的数据格式的文件；创建的文件必须在 dataDirectory 目录下，并通过自动移动或重命名成数据目录；文件一旦移动就不能被改变，如果文件被不断追加,新的数据将不会被阅读。对于简单的文本文，可以使用一个简单的方法streamingContext.textFileStream(dataDirectory)来读取数据。
Spark Streaming也可以基于自定义 Actors 的流创建DStream ，通过 Akka actors 接受数据流，使用方法streamingContext.actorStream(actorProps, actor-name)。
Spark Streaming使用 streamingContext.queueStream(queueOfRDDs)方法可以创建基于 RDD 队列的DStream，每个RDD 队列将被视为 DStream 中一块数据流进行加工处理。
2.2.2.2 高级来源
这一类的来源需要外部 non-Spark 库的接口，其中一些有复杂的依赖关系(如 Kafka、Flume)。因此通过这些来源创建 DStreams 需要明确其依赖。例如，如果想创建一个使用 Twitter tweets 的数据的DStream 流，必须按以下步骤来做：
1）在 SBT 或 Maven工程里添加 spark-streaming-twitter_2.10 依赖。
2）开发：导入 TwitterUtils 包，通过 TwitterUtils.createStream 方法创建一个DStream。
3）部署：添加所有依赖的 jar 包(包括依赖的spark-streaming-twitter_2.10 及其依赖)，然后部署应用程序。
需要注意的是，这些高级的来源一般在Spark Shell中不可用，因此基于这些高级来源的应用不能在Spark Shell中进行测试。如果你必须在Spark shell中使用它们，你需要下载相应的Maven工程的Jar依赖并添加到类路径中。
其中一些高级来源如下：
需要重申的一点是在开始编写自己的 SparkStreaming 程序之前，一定要将高级来源依赖的Jar添加到SBT 或 Maven 项目相应的artifact中。常见的输入源和其对应的Jar包如下图所示。
另外，输入DStream也可以创建自定义的数据源，需要做的就是实现一个用户定义的接收器。
2.2.3 DStream的操作
与RDD类似，DStream也提供了自己的一系列操作方法，这些操作可以分成三类：普通的转换操作、窗口转换操作和输出操作。
2.2.3.1 普通的转换操作
普通的转换操作如下表所示：
在上面列出的这些操作中，transform()方法和updateStateByKey()方法值得我们深入的探讨一下：
此函数应用于含有键值对的DStream中（如前面的示例中，在DStream中含有（word，1）键值对）。它会针对里面的每个元素（如wordCount中的word）调用一下更新函数，newValues是最新的值，runningCount是之前的值。
2.2.3.2 窗口转换操作
Spark Streaming 还提供了窗口的计算，它允许你通过滑动窗口对数据进行转换，窗口转换操作如下：
批处理间隔示意图
在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操作）。以下表列出了目前主要的输出操作：
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的基本特性。
类似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.优化运行时间
2.优化内存使用
参考资料：
（1）《Spark Streaming》 http://blog.debugo.com/spark-streaming/
版权声明：本文为博主原创文章，未经博主允许不得转载。
Spark入门实战系列--7.Spark Streaming（上）--实时流计算Spark Streaming介绍
原文地址：http://blog.csdn.net/yirenboy/article/details/48261695
