[大数据]Spark分析（十三）Spark Streaming中Spark 2.X的流计算

2021SC@SDUSC

前言

上一篇博客分析了Spark Streaming的性能调优机制，这次分析一下Spark 2.X的流计算过程

介绍

Spark 发展迅速，如今最新的版本已经是3.X了，但由于分析的需要以及减少分析困难，我们小组选定的是3.1.2的版本（详见小组环境配置博客：山东大学软件工程应用与实践——Spark项目（一）），但本次分析不针对Spark3.X的流计算特性而转而分析Spark2.X中的内容，是为了便于为Spark进行理解。这里也结合databricks官方演示文档里的一些图片进行分析。

Spark流计算

Spark2.X中相比于Spark1.6.X之前的版本，其提出了包含三个主题的更新：Easier、Faster、Smarter。
Spark2.X将流式计算也统一到DataFrame里中，提出了Structured Streaming的概念。功能更强大，效率更高，与其他组件的整合性也更好。
Structured Streaming的核心是将流式的数据看成一张不断增加的数据表，这种流式的数据处理模型类似于数据块处理模型，你可以把静态数据库表的一些查询操作应用在流式计算中，Spark运行这些标准的SQL查询，从不断增加的无边界表中获取数据。
不断输入的流式数据会被加载为内存中一张没有边界的数据表，每一条新来的数据都会作为一行数据新增到这张表中。

一、连续应用程序

首页，也是最重要的，在Spark2.X中提出了一个叫做连续应用程序（continuous application）的概念。
下图展示了一个围绕流数据展开的各种业务，数据从Kafka中流进来，通过ETL操作进行数据清洗，清洗出来作为目标数据进行进一步处理，可能是机器学习，也可能是交互式查询，也有可能直接把数据存在数据库或者其他外部存储设备，还有可能是直接交给已有的应用程序。

围绕流数据展开的多个端到端处理的连续应用程序

Spark 2.X把流数据看成是一个没有边际的表，并能把全部处理环节串联起来，形成端到端（end to end）处理。
而连续应用程序的模型就与这个类似，在充分应对风险的前提下，可以串联业务的全部过程。
下面的几行Scala代码就可以贯穿一个业务案例从始至终的业务流程。

logs = ctx.read.format("json").stream("s3://logs")

logs.groupBy("userid","hour").avg("latency").write.format("jdbc").startStream("jdbc:mysql//...")

二、无边界表unbounded table

对Spark Streaming来说连续（continuous）还有另一层含义，即运行在Dataset和DataFrame.之上。
基本观点是把数据看成一张表，默认情况下Dataset和DataFramer中的表是有边界的，而在流处理中是无边界的（unbounded）。对Spark Streaming来说，是将数据抽象为一个没有边界的表。没有了DStream，没有了先将数据收集过来再处理的概念。这个做法有一个非常大的好处。我们知道，目前Spark Streaming.直接依赖RDD，优化需要开发者自己完成，使用Dataset和DataFrame就可以利用Tungsten引擎来进行优化。默认情况下，Dataset、DataFrame是静态有边界数据（static bounded data），流数据是流式无边界数据（streaming unbounded data）。API把两者融合在一起，如下图所示。

Dataset/DataRrame API

如下图所示，新加入的Planner就类似路由器，我们在使用时，可以按照时间说明，由Planner确定每次读取的位置，在运行时动态绑定位置。在这种模式下，没有数据收集再处理的概念，可以认为数据一直在那儿，直接拿了处理就行。这可以极大地简化对流处理。

三、增量输出模式

在Spark 2.X中，增加了多个输出模式，增量输出（delta ouput） 是其中最重要的一种，如下图所示：

增量输出模式

增量更新，也就是说有需要更新的数据的才会更新，其他的不变。Trigger会不断检测输入数据，在不断地进行处理之后，输出结果只更新需要更新的内容，这个更符合应用程序的处理场景。

四、API简化

在API方面引入和流函数的封装。
这里举一个例子：Kafka中读取的数据，通过stream方法形成流，就可以直接与JDBC中读取的数据在Dataset层面进行join，不用transform或者foreachRDD方法。

kafkaDataset = spark.read.kafka("iot-updates").stream()

staticDataset = ctxt.read.jdbc("jdbc://", "iot-device-info")

joinedDataset = kafkaDataset.join(staticDataset, "device-type")

stream方法底层依赖Dataset和DataFrame，集成了Spark SQL和Dataset几乎所有的功能，把流处理的代码写一下简化了很多。

五、其他改进

Spark2.X同时也解决了DStream的很多问题：

1. 增加了eventTime的概念，把原有基于mini batch 处理的基础上，学习了Storm基于每个record的事件处理机制。
2. 可以把Spark Streaming抽象成一个数据库，直接通过JDBC访问数据。
3. 在运行时可以变更query，并支持多个query并行运行。

总结

从Spark2.X的设计来看，从根本上，是为了满足更快、完全容错、完全的语义一致性exactly-once的要求。通过实现由状态流处理，让应用程序的功能更强大。而基于Dataset和DataFrame处理，让我们忘记流的概念，使用将会越来越简单。