Apache SparkStreaming 简介和编程模型

短命女 2022-05-23 08:19 121阅读 0赞

## **1. 简介** ##

![70][]

图5.22 SparkStreaming\[16\]

Spark Streaming是Spark API核心扩展，提供对实时数据流进行流式处理，具备可扩展、高吞吐和容错等特性。Spark Streaming支持从多种数据源中提取数据，例如Twitter、Kafka、Flume、ZeroMQ和TCP套接字，并提供了一些高级的API来表示复杂处理算法，如map、reduce、join、windows等，最后可以将得到的结果存储到分布式文件系统（如HDFS）、数据库或者其他输出，Spark的机器学习和图计算的算法也可以应用于Spark Streaming的数据流中。Spark Streaming的本质实际上是一个微批处理系统，正因如此，Spark Streaming具有一些现有的流处理模型所没有的特性。它可以对故障节点和慢节点实现秒级的恢复，且具有高吞吐量。但其实时计算延迟是在秒级的，而现有的流处理系统（如Storm）一般是在毫秒级，所以Spark Streaming不适用于一些实时性要求很高的场景，如实时金融系统等。

许多数据需要实时进行处理，也就是说数据产生时的价值最大。例如，一个社交网络想在分钟级别内确定某个交流话题的趋势，搜索网站想根据用户的访问训练模型，服务商想在秒级内通过挖掘日志找到错误信息。设计适用于这些场景的模型极具挑战性，因为对于一些应用场景（如机器学习、实时日志分析），集群规模会达到百级以上，在这样的规模下会存在两个主要问题：节点故障（faults）和慢节点（slow nodes）问题。这两个问题在大规模集群下都是经常存在的，所以快速恢复在流系统应用中是十分重要的，否则流式应用可能无法及时做出关键的决定。但现有的一些流处理系统在这两个问题的处理上都十分有限，大多数流处理系统（如Storm、TimeStream、MapReduce Online等）都是基于纯实时的计算模型（a-record-at-a-time，来一条数据就处理一条数据），虽然这个模型能够有较小的计算时延，但是很难解决节点故障和慢节点的问题。一些传统的流式处理方法在小规模集群下运行较好，但在大规模情况下却面临着实质性的问题。

Spark Streaming提供了一种抽象的连续数据流，即Discretized Stream（离散流），一个离散流本质上就是一个序列化的RDD（Resilient Distributed Datasets，弹性分布式数据集）。离散流模型利用其并行恢复（ParallelRecovery）解决了节点故障和减轻了慢节点所带来的问题，还保证了一致性语义。

## **2. 系统架构** ##

离散流是Spark Streaming提供的基础抽象，它代表持续性的数据流，这些数据流既可以从外部源（如Kafka、Flume等）获取，也可以通过离散流的算子操作来获得。实质上，离散流由一组时间上连续的RDD组成，每个RDD都包含着一定时间片的数据，如图5.23所示：

![70 1][]

图5.23Discretized Stream\[17\]

![70 2][]

图5.24 SparkStreaming 整体架构\[18\]

如图5.24所示，这是Spark Streaming系统的整体架构，它将实时的流数据分解成一系列很小的批处理作业。批处理引擎使用的是Spark Core，也就是把输入数据按照一定的时间片（如1s）分成一段一段数据，每一段数据都会转换成Spark的RDD输入到Spark Core中，然后再将离散流的操作转换为RDD的算子操作，RDD算子操作产生的中间结果会保存在内存中，最后整个流式计算可以将中间结果输出到外部。

## **3. 一致性语义和容错** ##

对于流式计算，容错性的重要性在第一小节已经详细说明过了。首先，我们需要回忆Spark中RDD的容错机制。RDD是一个弹性不可变的分布式数据集，Spark记录着确定性的RDD转换的操作继承关系（lineage），所以只要输入的数据是可容错的，任何一个RDD的分区出错时，都可以根据lineage对原始输入数据进行转换操作，从而重新计算。图5.25是Spark Streaming的一个RDD继承关系图：

![70 3][]

图5.25 统计网页浏览量的lineragegraph\[18\]

图中每个椭圆代表的是一个RDD，椭圆中的每一个圆形是一个RDD的分区，图中的每一列的所有RDD代表的是一个离散流（图中一共有3个离散流），间隔\[0,1)和\[1,2)代表的是不同时间分片，图中每一行的最后一个RDD代表的是中间结果RDD。

并行恢复（Parallel Recovery）：系统会周期性的checkpoint RDD的数据，异步的备份到其他节点（默认复制数是2），因为RDD是不可变的，所以checkpoint不会锁住当前时间片的执行。一个Spark Streaming的流式应用，系统会每分钟对中间结果RDD进行checkpoint。当一个节点发生故障了，系统监测出丢失的RDD，系统会选择上一个checkpoint的数据来进行重新计算。离散流可以利用充分利用分区的并行性来达到更快的恢复速度：1）与批处理系统很相似的是，每个节点上运行多个task，每一个时间片的转换操作会在每个节点创建多个RDD分区（例如在100个节点的集群上有1000个RDD分区）。这样当一个节点发生故障时，可以让RDD的不同分区并行恢复。2）继承关系图（lineage graph）可以使不同时间片的数据并行恢复。如果一个机器节点发生故障，系统在每一个时间片可能丢失一些map操作的输出，从图5.26的浏览量统计应用的lineage graph可以看出，不同时间片的map可以并行地恢复计算，所以并行恢复的速度是比上游缓存策略更快的。

慢节点问题：在现有的纯实时流处理系统中，基本都没有解决慢节点的问题。离散流则与批处理系统类似，通过运行慢任务的副本来减轻慢节点带来的影响。Spark Streaming最开始采用一种简单的阈值来判断一个任务是否是慢任务：当一个任务是这个阶段（stage）的中间任务运行时间的1.4倍，则判断这是个慢任务。

一致性语义：离散流还有一个好处就是提供了强一致性。例如，考虑一个系统统计男女网页浏览量的比例，一个节点统计男性网页浏览量，另一个节点统计女性网页浏览量。如果一个节点落后于另一个节点，那么最终的结果也将有误。一些系统(如Borealis)利用同步节点来避免这个问题，而Storm就直接忽略了这个问题。而且Storm只能保证一个记录最少被处理一次，可能存在错误记录被多次处理，这就会使可变更的状态因更新两次而导致结果不正确，虽然Storm提供了Trident可以确保每条记录有且仅被处理一次，但是非常慢且需要用户去实现。使用离散流可以保证一致性是很明显的，因为时间被划分成时间片，每一个时间片的输出RDD都与这个时间片的输入和前面时间片有关(参考图5.26)，而RDD是不可变的，因此最终的结果是不会改变的。

## **4. Apache SparkStreaming编程模型** ##

### 4.1 数据模型 ###

在第2节我们知道，Spark Streaming就是把数据流划分为微批交给Spark Core处理的。Spark Core的处理的数据被抽象成了一个RDD，而Spark Streaming的处理数据被抽象成了一系列的DStream。实质上，离散流由一组时间上连续的RDD组成，每个RDD都包含着一定时间片的数据，如图5.23所示。

### 4.2 计算模型 ###

Spark Streaming的编程模型可以看成是一个批处理Spark Core的编程模型，除了API是调用Spark Streaming的API，很多概念都是一样的。在Spark Core编写程序时，只需要指定初始RDD的生成，然后对初始RDD进行一系列转换的操作，不断生成新的RDD，最后生成最终的结果RDD。

Spark Streaming也是类似的计算模型，DStream本质是一组时间上连续的RDD组成的，RDD是依靠着分区（Partition）来保证并行性的。在编写Spark Streaming程序的时候，我们需要指定初始DStream的输入源，生成初始的DStream，然后定义一些转换操作，这些DStream的操作最终都会转换成RDD的操作，然后在每一个时间片内，可以获得最终的结果DStream对应的RDD（也可以将结果选择输出到外部文件中），可以参考后面单词计数的实例分析。

PS：关于Spark Core中RDD的编程模型不属于本章所要讲的重点，在这里就不做赘述。

### 4.3 基本操作 ###

从Spark Streaming的系统架构可知，Spark Streaming中对DStream的各种操作，最终会在Spark Core中转换成RDD的操作，因此对DStream的操作是与Spark Core对RDD的操作是十分类似的。Spark Streaming在其数据模型DStream的模型下，为DStream提供了一系列的操作方法，这些操作大概可以分为3类：普通的转换操作、窗口转换操作和输出操作。常用的普通转换操作有flatMap、map、filter、reduceByKey、countByKey等操作，并且Spark Streaming支持将DStream的数据输出到外部系统，如数据库或文件系统。具体Spark Streaming支持的所有操作，可以到官网查看。

4.4 编程模型实例分析

下面用最基本的wordcount例子来解释其编程模型，其DStream的转换如下所示：

![70 4][]

图XXX：单词计数的DStream转换图

如上图所示，一共定义了四个离散流，wordCounts的离散流是我们最终要的结果。LinesDStream可以从文件系统、数据库、kafka等获取，然后对其进行flatMap操作，将每一行的文本分割成单词，形成新的离散流words DStream，随即进行mapToPair操作，将其映射成<word,1>的模式，最后用reduceByKey操作对每个单词进行计数，得到最终的结果离散流wordCountsDStream。

Java核心代码如下：

//创建SparkConf对象
    //与Spark Core的有一点不同，设置Master属性的时候，使用local模式时，
    // local后面必须跟一个方括号，里面填写一个数字，数字代表了用几个线程执行Spark Streaming程序。
    SparkConf conf = new SparkConf()
            .setMaster("local[2]")
            .setAppName("WordCountLocal");
    
    
    //创建SparkStreamingContext对象，还需指定每隔多长时间的数据划分为一个batch，这里是1s
    JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(1));
    
    //首先，创建一个DStream，代表了从一个数据源（这里是socket）来的持续不断的实时数据流
    JavaReceiverInputDStream<String> lines = jsc.socketTextStream("localhost", 9999);
    
    //将一行行的文本用flatMap切分成多个单词，words DStream的RDD元素类型为一个个单词
    JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
        @Override
        public Iterator<String> call(String line) throws Exception {
            return Arrays.asList(line.split(" ")).iterator();
        }
    });
    
    /
    //接着开始进行mapToPair操作，将单词映射成<word,1>的pair格式，得到离散流pairs
    JavaPairDStream<String,Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
        @Override
        public Tuple2<String, Integer> call(String word) throws Exception {
            return new Tuple2<String, Integer>(word,1);
        }
    });
    
    //对离散流pairs进行reduceByKey操作，进行单词计数，得到wordCounts离散流
    JavaPairDStream<String,Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer v1, Integer v2) throws Exception {
            return v1 + v2;
        }
    });
    //最后，每次计算完，就打印这一秒钟的单词计数情况
    wordCounts.print();
    //必须调用JavaStreamingContext的start()方法，整个Java Streaming Application才会启动执行
    //否则，不会执行
    jsc.start();
    try {
        jsc.awaitTermination();//等待应用程序的终止，可以使用CTRL+C手动停止
        //也可以通过调用JavaStreamingContext的stop()方法来终止程序
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    jsc.close();

[70]: /images/20220523/d75f8f4b3250475c8c417b59db799f4e.png
[70 1]: /images/20220523/2a70140a9f8c4f8e995d0882550e7be4.png
[70 2]: /images/20220523/c4c7b47865f24c6d9bc416d903594cde.png
[70 3]: /images/20220523/691a6448cbfe43a39dab6eb789fae149.png
[70 4]: /images/20220523/13c0d76921924112b5f4250fe7bad343.png