Spark Streaming

1. Spark Streaming介绍

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，然后进行处理，这是一个典型的生产者消费者模型，对应的就有生产者消费者模型的问题，即如何协调生产速率和消费速率。

2. Storm与Spark Streming比较

1、处理模型以及延迟

虽然两框架都提供了可扩展性(scalability)和可容错性(fault tolerance)，但是它们的处理模型从根本上说是不一样的。Storm可以实现亚秒级时延的处理，而每次只处理一条event，而Spark Streaming可以在一个短暂的时间窗口里面处理多条(batches)Event。所以说Storm可以实现亚秒级时延的处理，而Spark Streaming则有一定的时延。

2、容错和数据保证

然而两者的代价都是容错时候的数据保证，Spark Streaming的容错为有状态的计算提供了更好的支持。在Storm中，每条记录在系统的移动过程中都需要被标记跟踪，所以Storm只能保证每条记录最少被处理一次，但是允许从错误状态恢复时被处理多次。这就意味着可变更的状态可能被更新两次从而导致结果不正确。
另一方面，Spark Streaming仅仅需要在批处理级别对记录进行追踪，所以他能保证每个批处理记录仅仅被处理一次，即使是node节点挂掉。虽然说Storm的 Trident library可以保证一条记录被处理一次，但是它依赖于事务更新状态，而这个过程是很慢的，并且需要由用户去实现。

3、实现和编程API

Storm主要是由Clojure语言实现，Spark Streaming是由Scala实现。如果你想看看这两个框架是如何实现的或者你想自定义一些东西你就得记住这一点。Storm是由BackType和 Twitter开发，而Spark Streaming是在UC Berkeley开发的。
Storm提供了Java API，同时也支持其他语言的API。 Spark Streaming支持Scala和Java语言(其实也支持Python)。

4、批处理框架集成

Spark Streaming的一个很棒的特性就是它是在Spark框架上运行的。这样你就可以想使用其他批处理代码一样来写Spark Streaming程序，或者是在Spark中交互查询。这就减少了单独编写流批量处理程序和历史数据处理程序。

5、生产支持

Storm已经出现好多年了，而且自从2011年开始就在Twitter内部生产环境中使用，还有其他一些公司。而Spark Streaming是一个新的项目，并且在2013年仅仅被Sharethrough使用(据作者了解)。
Storm是 Hortonworks Hadoop数据平台中流处理的解决方案，而Spark Streaming出现在 MapR的分布式平台和Cloudera的企业数据平台中。除此之外，Databricks是为Spark提供技术支持的公司，包括了Spark Streaming。
虽然说两者都可以在各自的集群框架中运行，但是Storm可以在Mesos上运行, 而Spark Streaming可以在YARN和Mesos上运行。

3. Spark Streaming架构

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

1、计算流程：

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、容错性：

对于流式计算来说，容错性至关重要。首先我们要明确一下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中RDD的lineage关系图

3、实时性：

对于实时性的讨论，会牵涉到流式处理框架的应用场景。Spark Streaming将流式计算分解成多个Spark Job，对于每一段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程。对于目前版本的Spark Streaming而言，其最小的Batch Size的选取在0.5~2秒钟之间（Storm目前最小的延迟是100ms左右），所以Spark Streaming能够满足除对实时性要求非常高（如高频实时交易）之外的所有流式准实时计算场景。

4、扩展性与吞吐量：

Spark目前在EC2上已能够线性扩展到100个节点（每个节点4Core），可以以数秒的延迟处理6GB/s的数据量（60M records/s），其吞吐量也比流行的Storm高2～5倍，图4是Berkeley利用WordCount和Grep两个用例所做的测试，在Grep这个测试中，Spark Streaming中的每个节点的吞吐量是670k records/s，而Storm是115k records/s。

Spark Streaming与Storm吞吐量比较图

4. 操作示例

需要统计接收的文本数据中的每个单词的出现频率：

首先，我们将Spark Streaming相关的类和StreamingContext的一些隐式转换导入到我们的环境中，以便为我们需要的其他类（如DStream）添加有用的方法。StreamingContext是所有流功能的主要入口点。我们创建一个带有两个执行线程（译者注：如果要执行本例，必须确保机器cpu核心大于2）的本地StreamingContext，并且设置流数据每批的间隔为1秒。

import org.apache.spark._  
import org.apache.spark.streaming._  
import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3  
// 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))  

使用此context，我们可以创建一个DStream，它表示来自特定主机名（例如localhost）和端口（例如9999）TCP源的流数据。

// Create a DStream that will connect to hostname:port, like localhost:9999  
val lines = ssc.socketTextStream("localhost", 9999)  

在这行代码中，DStream表示从数据服务器接收的数据流。此DStream中的每个记录都是一行文本。接下来，我们要将每行文本以空格符为分隔符切分成一个个单词。

// Split each line into words  
val words = lines.flatMap(_.split(" "))  

flatMap是一个一对多的DStream操作，该操作通过从源DStream中的每个记录生成多个新记录来创建新的DStream。在这种情况下，每一行将被分割成多个单词，并将单词流表示为单词DStream。接下来，我们对这些单词进行计数。

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3  
// 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()  

单词DStream进一步映射（一对一变换）到(word,1) 键值对的DStream，然后进行聚合以获得每批数据中的单词的频率。最后，wordCounts.print()将打印每秒产生的计数结果中的若干条记录。

请注意，当执行这些代码时，Spark Streaming仅是设置了预计算流程，目前为止这些计算还没有真正的开始执行。在设置好所有计算操作后，要开始真正的执行过程，我们最终需要调用如下方法：

ssc.start()             // Start the computation  
ssc.awaitTermination()  // Wait for the computation to terminate  

完整的代码可以在SparkStreaming示例NetworkWordCount中找到。

如果您已经下载并构建了Spark，则可以以下面的方式运行此示例。在运行spark程序之前您将首先需要运行Netcat（大多数类Unix系统中的一个小型实用程序）作为数据服务器。

$ nc -lk 9999
然后，打开另外一个终端，键入一下命令启动示例

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
然后，在运行netcat服务器的终端中输入的任何行将每秒进行单词计数并打印在屏幕上。 

完整代码：

package org.apache.spark.streaming.examples  

import org.apache.spark.streaming.{Seconds, StreamingContext}  
import org.apache.spark.streaming.StreamingContext._  
import org.apache.spark.storage.StorageLevel  

object NetworkWordCount {  
  def main(args: Array[String]) {  
    if (args.length < 3) {  
      System.err.println("Usage: NetworkWordCount <master> <hostname> <port>\n" +  
        "In local mode, <master> should be 'local[n]' with n > 1")  
      System.exit(1)  
    }  

    StreamingExamples.setStreamingLogLevels()  

    // Create the context with a 1 second batch size  
    val ssc = new StreamingContext(args(0), "NetworkWordCount", Seconds(1),  
      System.getenv("SPARK_HOME"), StreamingContext.jarOfClass(this.getClass))  

    // Create a NetworkInputDStream on target ip:port and count the  
    // words in input stream of \n delimited text (eg. generated by 'nc')  
    val lines = ssc.socketTextStream(args(1), args(2).toInt, StorageLevel.MEMORY_ONLY_SER)  
    val words = lines.flatMap(_.split(" "))  
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)  
    wordCounts.print()  
    ssc.start()  
    ssc.awaitTermination()  
  }  
}