Streaming 一文读懂 Spark 和 Spark Streaming

前言

Apache Spark是最流行的开源大数据处理框架。和MapReduce一样，Spark用于分布式、大规模的数据处理，但Spark作为MapReduce的继承者，提供了更高级别的编程接口和更高的性能。此外，Spark不仅可以执行常规批处理计算，还可以提供流式计算支持。

阿帕奇火花(Apache Spark)诞生于著名的AMPLab(Meos和Alluxio也诞生于此)。它从一开始就有很强的学术气质，其设计目标是为各种大数据处理需求提供统一的技术栈。目前，Spark背后的商业公司Databricks的创始人也是AMPLab的博士毕业生。

Spark本身是用Scala语言写的。Scala是一种面向对象和函数相结合的“双范式”语言，运行在JVM上。Spark广泛使用其函数表达式、即时代码生成等功能。Spark目前提供Java、Scala、Python和R中的API，前两者因为也运行在JVM上，所以可以实现更多的原生支持。

MapReduce有什么问题

Hadoop是大数据处理领域的先驱。严格来说，Hadoop不仅仅是一个软件，而是一个完整的生态系统。例如，MapReduce负责分布式计算，而HDFS负责存储大量文件。

MapReduce模型的诞生是大数据处理从无到有的飞跃。然而，随着技术的发展，对大数据处理的需求越来越复杂，MapReduce的问题日益突出。通常，我们将MapReduce的输入和输出数据保存在HDFS。在很多情况下，复杂的ETL、数据清理等工作无法用MapReduce一次完成，因此我们需要连接多个MapReduce流程:

▲上图中，只有两个MapReduce串联。其实可能有几十个甚至更多，依赖关系更复杂。

这样，每一个中间结果都必须写入HDFS存储，成本很高(别忘了，HDFS的每一个数据都需要冗余拷贝)。另外，因为本质上是多个MapReduce任务，调度也比较麻烦，实时性无从谈起。

斯帕克和RDD模型

要解决以上问题，如果能把中间结果保存在内存里岂不是要快很多？最大的障碍是分布式系统必须能够容忍某些故障，称为容错。如果只存储在内存中，一旦某个计算节点宕机，其他节点就无法恢复丢失的数据，只能重启整个计算任务，这对于节点数以百计的集群来说是无法接受的。

一般来说，实现容错的方法只有两种:要么将其存储在外部(如HDFS)，要么将其复制到多个副本。Spark大胆提出了第三种——重新计算。但是，这取决于一个附加的假设:所有的计算过程都是确定性的。Spark借用了函数式编程的思想，提出了RDD(弹性分布式数据集)，被翻译成“弹性分布式数据集”。

RDD是一个只读的分区数据集。RDD要么来自不可变的外部文件(如HDFS的文件)，要么由其他RDD的某些操作者计算。RDD由算子连接形成有向无环图。上图显示了一个简单的例子，其中节点对应于RDD，边对应于运算符。

回到刚才的问题，RDD如何实现容错？很简单，RDD的每个分区都可以确定性地计算，所以一旦一个分区丢失，另一个计算节点可以从它的前一个节点开始，用相同的计算过程重新计算一次，从而获得完全相同的RDD分区。这个过程可以递归执行。

▲上图为RDD分区的恢复。为了简洁起见，没有画分区。其实恢复是基于分区的。

Spark的编程接口非常类似于Java 8的Stream: RDD作为数据，在各种操作符之间进行转换，形成执行计划DAG的描述。最后，一旦遇到类似collect()的输出命令，执行计划将被发送到Spark集群开始计算。不难发现，运营商分为两类:

map()、filter()、join() 等算子称为 Transformation，它们输入一个或多个 RDD，输出一个 RDD。collect()、count()、save() 等算子称为 Action，它们通常是将数据收集起来返回；

▲以上示例用于收集包含关键字“HDFS”的错误日志时间戳。当执行到collect()时，右边的执行计划开始运行。

如前所述，RDD的数据由多个分区组成，可以分布在集群中的各种机器上，这就是RDD“分布式”的含义。熟悉DBMS的同学可以把RDD理解为逻辑执行计划，把分区理解为物理执行计划。

此外，RDD还包含每个分区的依赖关系，一个函数指示如何计算该分区的数据。Spark设计师发现，依赖关系自然可以分为两种:窄依赖关系和宽依赖关系，例如:

map()、filter() 等算子构成窄依赖：生产的每个分区只依赖父 RDD 中的一个分区。groupByKey() 等算子构成宽依赖：生成的每个分区依赖父 RDD 中的多个分区（往往是全部分区）。

▲左图为宽依赖和窄依赖，其中Join运算符由于Join键的划分条件不同，两者都有；右图显示了执行过程。由于广泛依赖性的存在，执行计划分为三个阶段。

在执行的时候，可以很容易的用流水线的方式计算出狭窄的依赖关系:对于每个分区，从前到后依次替换每个操作符。但是，宽相关性需要等待计算前一个RDD中的所有分区。换句话说，广泛的依赖性就像一个屏障，阻挡了所有以前的计算。整个计算过程被广泛的依赖分为多个阶段，如上图所示。

了解MapReduce的同学可能发现，广泛依赖本质上是一个MapReduce过程。但是相对于MapReduce自己编写Map和Reduce函数的编程接口，Spark的接口要容易得多；而在Spark中，多阶段的MapReduce只需要构造一个DAG。

声明式接口：Spark SQL

Spark诞生后，MapReduce编程模型大大简化，但人们并不满意。众所周知，与命令式编程相反，它是声明式编程。前者需要告诉程序如何得到我需要的结果，后者告诉程序我需要的结果是什么。比如你想知道，每个部门

在命令式编程中，你需要写一个程序。以下是伪代码实现:

employees = db . getallemployees()= countby dept = dict()//count of employee in employee:if(employee . gender = = ' employee ')countby dept[employee . dept _ id]+= 1 results = list()//Add dept.name column dept = db . getall departments()for dept in dept:if(countby dept包含key dept . id)results . Add(row(dept . id，dept . name，countbydept [dept.id])返回结果；

在声明式编程中，您只需要用关系代数运算来表示结果:

employees.join(dept，employees.deptId == dept.id)。其中(employees.gender == 'female ')。groupBy(dept.id，dept.name)。agg()

等价地，如果你对SQL比较熟悉，也可以这样写:

SELECTdept.id，dept.name，COUNT(*)from Employees join dept one employees . dept _ id = = dept . id where Employees . gender = ' meeting ' group by dept . id，dept.name

显然，声明式要简单得多！但是声明式编程依赖于执行器生成的真实程序代码，所以除了上述程序之外，还需要将数据模型告知执行器。最著名的声明式编程形式是SQL。

Spark SQL就是这样一个基于SQL的声明式编程接口。你可以把它想象成Spark上面的一层封装。基于RDD计算模型，它提供了数据框架应用编程接口和内置的SQL执行计划优化器催化剂。

▲上图黄色部分是Spark SQL中的新部分。

数据框就像数据库中的一个表，它保存了数据之外的数据的模式信息。在计算中，模式信息将通过运算符进行转换。数据帧的数据是由行对象组成的RDD，对数据帧的操作最终会变成对底层RDD的操作。

Catalyst是一个内置的SQL优化器，负责将用户输入的SQL转换成执行计划。Catelyst的优点是利用Scala提供的codegen机制，编译物理执行计划，生成的执行代码非常高效，几乎和直接操作RDD的命令式代码一样。

▲上图为Catalyst的工作流程。像大多数SQL优化器一样，它是一个基于成本的优化器(CBO)，但它最终通过使用codegen转换成了在RDD的直接操作。

流媒体计算框架:星火流媒体

过去，批处理和流计算被认为是大数据系统的两个方面。我们经常可以看到这样的架构——以卡夫卡和风暴为代表的流媒体计算框架用于实时计算，而Spark或者MapReduce则负责每天每小时的数据批处理。在ETL等场合，这种设计往往导致同一个计算逻辑被实现两次，既消耗人力，又保证一致性。

星火流媒体就是在这种需求下诞生的。传统的流计算框架大多关注低延迟，采用连续算子模型。Spark Streaming，在Spark的基础上，提出了一种新的D-Stream (Discarded Streams)方案:将流数据切割成微批量，使用一系列短的、无状态的、确定性的批量来实现流处理。

Spark Streaming在流式计算框架方面极具创新性。虽然牺牲了低延迟(一般流计算可以达到100ms，Spark Streaming延迟一般在1s左右)，但它带来了三个吸引人的优势:

更高的吞吐量（大约是 Storm 的 2-5 倍）更快速的失败恢复（通常只要 1-2s），因此对于 straggler（性能拖后腿的节点）直接杀掉即可开发者只需要维护一套 ETL 逻辑即可同时用于批处理和流计算

▲左上图，为了保证持久算子模型的流量计算系统的一致性，主备机之间不得不使用同步机制，导致性能损失。星火流媒体完全没有这个问题；右边是D-Stream原理示意图。

你可能会困惑。流计算中的状态一直是个问题。但是我们刚才提到D-Stream方案是无状态的，那么对于字数等问题，如何保持计数运算符的状态呢？

答案是通过RDD:把上一个时间步长的RDD作为当前时间步长的RDD的前身节点，可以带来不断改变状态的效果。事实上，新的RDD国总是会产生的，而旧的RDD不会被取代，而是会被用作新RDD的前身。对于底层的Spark框架，没有时间步长的概念，只有扩展的DAG图和新的RDD节点。

▲上图是流式字数统计的例子，统计结果是按不同的时间步长累加的。

那么另一个问题随之而来:随着时间的推移，上图中RDD计数的州会越来越多，他的世系也会越来越长。在极端情况下，恢复过程可能会追溯到很久以前。这是不能接受的！因此，星火Streming将定期检查RDD州，并将其保存到HDFS和其他储存地，这被称为血统切割，更早的RDD可以在它之前毫无顾虑地被清除。

关于几种流行的开源流计算框架的比较，请参考文章Apache流处理框架的比较。

流计算与 SQL：Spark Structured Streaming

Spark通过Spark Streaming拥有流计算能力，那么Spark SQL是否可以拥有类似的流处理能力？答案是肯定的，只要把数据流建模成一个没有边界的生长表，在这样的语义下，很多SQL操作都可以直接应用到流数据上。

没想到，Spark结构化流的流计算引擎并没有重用Spark Streaming，而是在Spark SQL上设计了一套新的引擎。因此，从Spark SQL迁移到Spark结构化流很容易，但从Spark Streaming迁移要困难得多。

自然，基于这个模型，Spark SQL中的大多数接口和实现都可以在Spark结构化流中直接重用。将用户的SQL执行计划转化为流计算执行计划的过程称为增量，由Spark框架自动完成。对于用户来说，只需要知道每次计算的输入是某个短时间段的流数据，输出是对应数据的计算结果。

▲左图为Spark结构化流模型示意图；右图显示了同一任务的批量和流量计算版本。可以看出，除了输入输出不同，内部计算过程完全一样。

与Spark SQL相比，流SQL计算有两个额外的特性，即窗口和水印。

一个窗口是过去某段时间的定义。在批处理中，查询通常是满的(比如用户总数是多少)；在流量计算中，我们通常关心的是最近的数据(比如最近24小时新增了多少用户)。用户可以通过选择合适的窗口，如滑动窗口、滚动窗口等，得到自己需要的计算结果。

水印用于丢弃过早的数据。在流计算中，上游输入事件可能存在不确定的延迟，但流计算系统的内存是有限的，只能保存有限的状态，一定时间后必须丢弃历史数据。以双流A JOIN B为例，假设窗口为1小时，A中比当前时间早1小时的数据(行)将被丢弃；如果B中有1小时前的事件，只能忽略，因为无法处理。

▲上图为水位示意图。太晚的数据(行)将被忽略，因为它们低于当前水位，无法处理。

水位和窗的概念来源于时间。在其他流计算系统中，有相同或相似的概念。

在SQL的流计算模型上，经常比较另一种流计算框架Apache Flink。和Spark相比，他们的实现思路大相径庭，但模型却非常相似。

系统架构

Spark中有三个角色:驱动程序、工作人员和集群管理器。

驱动程序是用户编写的程序，对应于一个SparkContext，负责任务构造、调度、故障恢复等。驱动可以直接在客户端运行，比如用户的应用；它也可以托管在主机上，这被称为集群模式，通常用于长期任务，如流计算。

顾名思义，集群管理器负责集群的资源分配。Spark自带的Spark Master支持任务的资源分配，并包含一个Web UI来监控任务的运行状态。多个主机可以组成一个主机和多个备用主机，ZooKeeper可以用于协调和故障恢复。星火主机一般可以用于星火集群。但是如果用户的集群不仅有Spark框架还承担其他任务，那么官方推荐Mesos作为集群调度器。

工作者节点负责执行计算任务，并存储RDD等数据。

总结

Spark是一个分布式计算系统，支持批处理和流计算。所有斯巴克的计算都是建立在RDD的基础上的，它是由操作员连接起来形成DAG的执行计划的。RDD的确定性和不变性是斯巴克故障恢复的基础。本质上，星火流的数据流也是一个RDD，它将输入数据分成微批量。

Spark SQL是RDD之上的一层封装。与原RDD相比，数据框架应用编程接口支持数据表的模式信息，可以执行SQL关系查询，大大降低了开发成本。Spark结构化流是Spark SQL的流计算版本，它将输入数据流视为一个不断增加的数据行。