流处理基础

Apache Spark一直支持流处理。在2012年，该项目合并了Spark Streaming
及其DStream API，这是第一个能够使用例如map和reduce这样的高级函数进行流处理
的API。现在，数以百计的组织在生产中使用DStreams来实现大型实时应用程序，通
常每小时处理TB级的数据。虽然在很多方面与弹性分布式数据集(RDD) API相似，但
是Dstream API是基于Java/Python对象的相对较低级别操作，它限制了更高级别的优
化。因此，在2016年，Spark项目增加了结构化流处理（Structured Streaming），这是
一个直接建立在DataFrame之上的新式流处理API，支持多种优化策略，并且可以很简
单地与其他DataFrame和Dataset代码集成在一起。在Apache Spark 2.2中，结构化流处
理API（Structured Streaming API）已经很稳定，并且在Spark社区中也得到了迅速地
应用。

---

提示：更多内容查看 http://faculty.neu.edu.cn/cc/zhangyf/book/spark.pdf

---

前言

在本书中，我们将只关注结构化流处理API，它直接与我们在本书前面讨论的
DataFrame和Dataset API集成在一起，同时它也是编写新式流处理应用程序的一种框
架。如果你对DStream感兴趣，许多其他的书籍涵盖了这个API的功能介绍，包括一
些关于Spark流处理的专门书籍，如Francois Garillot和Gerard Maas的《Learning Spark
Streaming》(O’Reilly，2017)。然而，与RDD和DataFrame一样，本书重点介绍的结构
化流处理包括了DStream的大多数功能，并且由于代码生成和Catalyst优化器，通常会
表现更好。

---

在讨论Spark的流处理API之前，我们首先形式化地定义流处理和批处理。本章将讨论
这一领域中的一些核心概念，虽然不能涵盖该领域的全部内容，但你将了解大部分概
念，这些足够让你能够理解这个领域的系统。

一、什么是流处理？

流处理是连续处理新到来的数据以更新计算结果的行为。在流处理中，输入数据是无边界的，没有预定的开始或结束。它是一系列到达流处理系统的事件（例如，信用
卡交易，单击网站动作，或从物联网IoT传感器读取的数据），用户应用程序对此事
件流可以执行各种查询操作（例如，跟踪每种事件类型的发生次数，或将这些事件按
照某时间窗口聚合）。应用程序在运行时将输出多个版本的结果，或者在某外部系统
（如键值存储）中持续保持最新的聚合结果。

当然，我们可以将流处理与批处理（batch processing）进行比较，批处理是在固定大
小输入数据集上进行计算的。通常，这可能是数据仓库中的大规模数据集，其包含来
自应用程序的所有历史事件（例如，过去一个月的所有网站访问记录或传感器记录的
数据）。批处理也可以进行查询计算，与流处理差不多，但只计算一次结果。

虽然流处理和批处理是不同的，但在实践中他们经常需要一起使用。例如，流式应用
程序通常需要将输入流数据与批处理作业定期产生的数据连接起来，流式作业的输出
通常是在批处理作业中要查询的文件或数据表。此外，应用程序中的任何业务逻辑都
需要在流处理和批处理执行之间保持一致： 例如，如果你有一个自定义代码来计算用
户的计费金额，而流处理和批处理运行出来的结果不同那就出问题了!为了满足这些
需求，结构化流处理从一开始就设计成可以轻松地与其他Spark组件进行交互，包括
批处理应用程序。结构化流处理开发人员提出了一个叫做连续应用程序（continuous
application）的概念，它把包括流处理、批处理和交互式作业等全部作用在同一数据
集上的处理环节串联起来，从而构建端到端应用程序 ，提供最终的处理结果或产品。
结构化流处理注重使用简单的端到端的方式构建此类应用程序，而不是仅仅在流数据
级别上针对每条记录进行处理。

二、流处理应用场景

我们将流处理定义为在无边界数据集上的增量处理，但显然这并不能直观的体现流处
理的应用场景，也不能体现其必要性。在我们介绍流处理的优缺点之前，先来了解一
下为什么你需要使用流处理，我们将描述来自底层流处理系统中具有不同需求的六个
常见应用场景。

1.通知和警报

可能最明显的流处理应用场景是通知和警报。给定一系列事件，如果发生某一事件或
一系列连续特殊事件，则应触发通知或警报。这不一定意味着触发自动决策或者触发
预先制定的决策，警报还可以通知人们以采取一些必要的行动。比如，流处理系统向
仓库管理员发出警报，提示他们需要从仓库中的某个位置取出货物，并将其装运给客
户，在任何情况下，这类通知都需要及时。

2.实时报告

许多组织使用流处理系统来运行员工可以实时查看的动态仪表板。例如，这本书的作者每天都要使用Databricks 公司（本书两位作者都受雇于该公司）的实时报告仪表
板，该仪表盘即利用结构化流处理运行。我们使用这些仪表板来监视整个平台的使
用情况、系统负载、在线时间，甚至是新功能在推出时以及其他应用程序中的使用情
况。

3.增量ETL

最常见的流处理应用程序之一是减少公司在信息检索时必须忍受的延迟时间，简而
言之，“把批处理任务用流处理方式执行”。Spark批处理作业通常用于抽取-转换加载任务（ETL任务，Extract-Transform-Load），将原始数据转换为结构化格式 (如
Parquet) 以支持高效查询。使用结构化流处理，这些作业可以在几秒钟内处理新数
据，使下游用户能够更快地查询信息。在这个应用场景中，关键是一次性地以容错的
方式处理数据： 我们不希望它在加载到数据仓库之前丢失任何输入数据，而且我们
不想两次加载相同的数据。此外，流处理系统需要支持数据仓库的事务级别的更新操
作，以避免查询结果与部分写入数据的冲突。

4.实时数据更新来提供实时服务

流处理系统经常用于为另一个应用程序提供实时的数据处理结果。例如，类似谷歌分
析（Google Analytics）这样的Web分析产品可能会持续跟踪每个网页的访问次数，并
使用流处理系统将这些计数持续更新，当用户与产品UI交互时，此Web应用程序将查
询最新的计数。支持此用例需要流处理系统对 key–value 存储 (或其他服务系统)执行
同步的增量更新，而且通常与ETL案例中的一样，这些更新需要是事务性的以保证应
用程序的数据一致性。

5.实时决策

流处理系统的实时决策包括分析新的输入，根据业务逻辑自动作出决策来应对新数
据。比如，一个银行希望自动验证客户信用卡上的新交易记录是否为一个可疑交易，
这种判断可以基于其最近的历史记录，如果认定可疑的盗刷行为则拒绝交易。此决
策需要在处理每个交易时实时进行，因此开发人员需要在流处理系统中实现此业务逻
辑，并用于处理交易信息的数据流。这种类型的应用程序需要维护每个用户的大量状
态，以跟踪其当前的支出模式，并自动将这些状态与每个新交易进行比较。

6.在线机器学习

与实时决策用例相近的是在线机器学习。在这种情况下，你可能希望基于多个用户的
流和历史数据的组合来训练模型。这个示例可能比前面提到的信用卡交易用例更复
杂，因为公司可能希望基于所有客户行为来不断更新模型并基于它验证每笔交易，而
不是去根据一个客户的行为采用固定规则来做出响应。这是流处理系统中最具挑战性
的应用场景，因为它需要对多个客户的数据进行聚合操作、与静态数据集执行连接操
作、与机器学习库集成等，并同时需要降低延迟响应时间。

以上是流处理的应用场景

---

流处理的优点

我们已经了解了一些使用流处理的应用场景，下面具体介绍一下流处理的优点。在大
多数情况下，批处理更容易理解、更容易调试、也更容易编写应用程序。此外，批量
处理数据也使得数据处理的吞吐量大大高于许多流处理系统。然而，流处理对于以下
两种情况非常必要。首先，流处理可以降低延迟时间：当你的应用程序需要快速响应
时间（在分钟、秒或毫秒级别上），你需要一个可以将状态保存在内存中的流处理系
统，以获得更好的性能，我们介绍的许多决策和警报案例都属于这种情况。其次，流
处理在更新结果方面也比重复的批处理作业更有效，因为它会自动增量计算。例如，
如果我们要计算过去24小时内的 Web 流量统计数据，那么简单的批处理作业实现可
能会在每次运行时遍历所有数据，总是处理24小时内的数据。与此相反，流处理系统
可以记住以前计算的状态，只计算新数据。如果你告诉流处理系统个小时更新一次报
告，则每次只需处理1小时的数据 (自上次报告以来的新数据)。在批理系统中，你需
要手动实现这种增量计算，以获得相同的性能，从而导致要做大量额外的工作，而这
些工作在流处理系统中会自动完成。

流处理的挑战：

我们讨论了流处理的应用场景和优点，但你可能知道，从来没有免费的午餐。让我们
讨论一下流处理的一些挑战。
为了更好地说明，我们举个实际例子。假设我们的应用程序接收来自传感器（例如，
流处理基础 ｜ 333
汽车内部）的输入消息，该传感器在不同时间报告其值。然后，我们希望在该数据流
中搜索特定值或特定模式，一个挑战就是输入记录可能会无序地到达应用程序。例
如，因延迟和重新传输，我们可能会收到以下顺序的更新序列，其中time字段显示实
际测量的时间：
{value: 1, time: “2017-04-07T00:00:00”}
{value: 2, time: “2017-04-07T01:00:00”}
{value: 5, time: “2017-04-07T02:00:00”}
{value: 10, time: “2017-04-07T01:30:00”}
{value: 7, time: “2017-04-07T03:00:00”}
在任何数据处理系统中，我们都可以构造逻辑来执行一些基于接收单值“5”的操
作。在流处理系统中，我们也可以快速响应此单个事件。然而，如果你只想根据收到
的特定值序列触发某个动作，比如2，然后10，接着5，事情就复杂多了。在批处理
的情况下，这并不特别困难，因为我们可以简单地按time字段对所有事件进行排序，
以发现10在2和5之间到来。然而，对于流处理系统来说，这是比较困难的。原因是
流处理系统将单独接收每个事件，并且需要跟踪事件的某些状态以记住值为2和5的事
件，并意识到值为10的事件是在它们之间，这种在流中记住事件状态的需求带来了更
多的挑战。例如，如果你有一个海量的数据集(例如，数以百万计的传感器流) 并且状
态本身是巨大的，应该如何处理？如果系统中的机器出现故障了，丢失了一些状态怎
么办？如果负载不平衡，一台机器的速度非常慢怎么办？当对某些事件的分析完成之
后，你的应用程序如何向下游消费者发出信号 (例如，模式 2-10-5没有发生)？是否应
该等待固定的时间或无限期地记住某个状态？当你希望部署流式应用程序时，所有这
些挑战和其他问题 (例如处理事务性的输入和输出) 都会出现。

总结一下，我们在前面几段所述的挑战如下：

1、 基于应用程序时间戳(也称为事件时间) 处理无序数据。
2、 维持大量的状态。
3、 支持高吞吐量。
4、 即使有机器故障也仅需对事件进行一次处理。
5、 处理负载不平衡和拖延者（straggler）。
6、快速响应时间。
7、与其他存储系统中的数据连接。
8、确定在新事件到达时如何更新输出。
9、事务性地向输出系统写入数据。
10、在运行时更新应用程序的业务逻辑。

这些主题都是大型流处理系统研究和开发的热点问题。为了了解不同的流处理系统如
何应对这些挑战，我们将介绍一些最常见的设计概念。

---

流处理设计要点：

为了应对我们所描述的流处理难题，包括高吞吐量、低延迟时间和无序数据处理，有
多种方法来设计流处理系统。在下一节中描述Spark的选择之前，我们在这里描述最
常见的设计选项。

记录级别API与声明式API：

设计流处理API 最简单的方法就是将每个事件传递给应用程序，并使用自定义代码
进行响应。这是许多早期的流处理系统 (如 Apache Storm) 的实现方法，当应用程序
需要完全控制数据处理时，它具有重要的地位。提供这种记录级别API（record-at-atime API）的流处理系统只是给用户提供一个获取每条流数据记录的接口，然而这些
系统的缺点是，我们前面描述的大多数复杂因素 (如状态维护) 都要完全由应用程序负
责。例如，使用记录级别API，用户需要负责在较长时间段内跟踪状态，在一段时间
后删除它以清除空间，并在发生故障后以不同方式响应重复事件。在这些系统上实现
正确的流处理程序相当困难。从本质上讲，像记录级别API这种低级 API 需要深厚的
专业知识才能开发和维护。
因此，许多后来出现的流处理系统提供了声明式API，应用程序为了响应每个新事
件指定要计算的内容，而不是如何计算，也不需要考虑如何从失败中恢复。例如，
Spark的原始 DStream API 提供了基于map、reduce和filter等数据流上操作的功能性
API。在内部，DStream API自动跟踪每个操作处理的数据量，可靠地保存相关状态，
并在需要时从失败中恢复计算，诸如 Google DataFlow和 Apache Kafka Stream等系统
也提供了类似的功能性API。Spark的结构化流处理则更进一步，从功能操作切换到关
系操作(类似 SQL)，从而在不需要编程的情况下实现更丰富的自动执行优化。

基于事件时间与基于处理时间：

对于具有声明性 API 的系统，第二个问题是系统是否支持事件时间。事件时间（event time）是根据数据源插入记录中的时间戳来处理数据的概念，而不是流处理应用程序，在接收记录时的时间（称为处理时间，processing time）。特别是，当使用事件时间时，记录可能会出现无序状况(例如，如果记录从不同的网络路径返回)，并且不同的数据源可能也无法同步（对于标记相同事件时间的记录，某些记录可能比其他记录晚到达）。如果你的应用程序从可能产生延迟的远程数据源 (如移动电话或物联网设备)收集数据，则基于事件时间的处理方式就非常必要。如果不基于事件时间，你可能在某些数据延迟到达时无法发现某些重要的模式。相比之下，如果应用程序只处理本地事件 (例如，在同一个数据中心中生成的数据)，则可能不需要复杂的事件时间处理。
在使用事件时间时，有几个问题需要重点关注，包括以允许系统合并延迟事件的方式跟踪状态，以及确定何时在事件时间内输出给定时间窗口的结果是安全的（即当系统已收到某事件时间点之前的所有输入）。因此，许多声明性系统（包括结构化流处理）对其所有API中集成的事件时间都具有“原生”支持，因此可以在整个程序中自动处理这些问题。

连续处理与微批量处理

最后一个设计选择通常是确定连续处理（Continuous Processing）模式还是微批量处
理（Micro-batch Processing）模式。在基于连续处理的系统中，系统中的每个节点都
不断地侦听来自其他节点的消息并将新的更新输出到其子节点。例如，假设你的应用
程序在多个输入流上实现了map-reduce运算。在连续处理系统中，实现map的每个节
点将从输入数据源一个一个地读取记录，根据函数逻辑执行计算，并将它们发送到相
应的reducer。当reducer获得新记录时，将更新其状态。最关键的是这种情况是发生在
每个记录上的，如图 20-1所示。

在输入数据速率较低时，连续处理提供尽可能低的处理延迟，这是连续处理的优势，
因为每个节点都会立即响应新消息。但是，连续处理系统通常具有较低的最大吞吐
量，因为它们处理每条记录的开销都很大 (例如，调用操作系统将数据包发送到下游
节点)。此外，连续处理系统通常具有固定的计算拓扑，如果不停止整个系统，在运
行状态下是不能够改动的，这也可能会导致负载均衡的问题。
相比之下，微批量处理系统等待积累少量的输入数据（比如500ms的数据），然后使
用分布式任务集合并行处理每个批次，类似于在Spark中执行批处理作业。微批量处
理系统通常可以在每个节点上实现高吞吐量，因为它们利用与批处理系统相同的优化
操作 (例如，向量化处理，Vectorized Processing)，并且不会引入频繁处理单个记录的
额外开销，如图 20-2所示。

在这两种执行模式之间进行选择时，最主要因素是你期望的延迟和总的操作成本
(TCO)。根据应用程序的不同，微批量处理系统可能将延迟从100毫秒延长到秒。在这
种机制下，通常需要较少的节点就可以达到要求的吞吐量，因而降低了运营成本 (包
括由于节点故障次数较少带来的维护成本降低)。为了获得更低的延迟，你应该考虑
一个连续处理系统，或使用一个微批量处理系统与快速服务层（fast serving layer）结
合以提供低延迟查询（例如，将数据输出到 MySQL 或 Apache Cassandra，可以在几
毫秒内将其提供给客户端）。

Spark的流处理API

我们已经讨论了流处理的一些高级设计方法，但到目前为止，我们还没有详细讨论
Spark的流处理 API。正如在本章开头说的那样，Spark包括两种流处理 API，Spark早
期的 DStream API 纯粹是基于微批量处理模式的，它是声明式 (基于功能的) API，但
不支持事件时间。新的结构化流处理 API添加了更高级别的优化、事件时间，并且支
持连续处理。

DStream API

Spark原始的DStream API自2012年首次发布以来已被广泛应用。例如，在Datanami
2016 调查结果中DStream是使用最广泛的流处理引擎。由于高级 的API 接口和简单的
语义，许多公司在大规模实际生产中都采用Spark流处理。与 RDD 代码的交互（如与
静态数据的连接）也是Spark流处理原生支持的。运行Spark流处理并不比运行正常的
Spark作业困难。但是，DStream API 有几个限制。首先，它完全基于Java/Python 对
象和函数，而不是DataFrame和Dataset中丰富的结构化表概念，这不利于执行引擎进
行优化。其次，DStream API 纯粹基于处理时间（要处理事件时间操作，应用程序需
要自己实现它们）。最后，DStream只能以微批量方式运行，并在其API的某些部分
暴露了微批量的持续时间，这使得其难以支持其他执行模式。

结构化流处理

结构化流处理是基于Spark结构化API的高级流处理 API，它适用于运行结构化处理的所有环境，例如Scala，Java，Python，R 和 SQL。与DStream一样，它是基于高级操作的声明式 API，但是，通过构建本书上一部分介绍的结构化数据模型，结构化流处理可以自动执行更多类型的优化。但是，与 DStream不同，结构化流处理对标记事件时间的数据具有原生支持 (所有窗口操作都自动支持它)。在Apache Spark 2.2中，系统只支持微批量模型，但Databricks 的Spark团队正在扩展对连续处理的支持，最终将支持连续处理的执行模式，这应该在Spark 2.3 版本上支持。更重要的是，除了简化流处理之外，结构化流处理还旨在使用Apache Spark轻松构建端到端连续应用程序，这些应用程序结合了流处理、批处理和交互式查询。例如，结构化流处理不使用 DataFrame之外的 API，只需编写一个正常的 DataFrame (或 SQL)计算并在数据流上应用它。当数据到达时，结构化流处理将以增量方式自动更新此计算的结果。这非常有益于简化编写端到端数据处理应用程序，开发人员不需要维护批处理代码版本和流处理版本，并且避免这两个版本代码失去同步的风险。另外一个例子是，结构化流处理可以将数据输出到由Spark SQL (如Parquet表) 可用的标准接收器，从而便于从另一个Spark应用程序查询你的流状态。在未来版本的 Apache Spark 中，我们期待越来越多的项目组件与结构化流处理集成进来，包括 MLlib 中的在线学习算法。总的来看，结构化流处理易于使用且更加高效，它是Spark流处理DStream API 的演化，因此我们将只关注更新的结构化流处理API。许多概念，例如从转换图中生成计算，也适用于 DStream，但我们不在本书中介绍，你可以参考其他书籍。

小结

本章介绍了流处理的基本概念和思路，本章介绍的设计方法帮助读者了解如何针对给
定应用程序设计流处理系统。你也应该理解了DStream和结构化流处理的作者做了哪
些权衡，以及使用结构化流处理时直接支持 DataFrame为什么如此重要，因为它避免
了重写应用程序逻辑。在接下来的章节中，我们将深入介绍结构化流处理，了解如何
使用它。

Spark 权威指南 第20章 流处理基础相关推荐

1. CUDA C编程权威指南 第六章 流和并发

   流 cuda流 流分为两种类型:1)隐式声明流(空流),2)显式声明流(非空流) cudaError_t cudaMemcpyAsync(void* dst,const void* src, size ...

2. Asterisk权威指南/第六章 拨号计划基础

   拨号计划是你的Asterisk系统的心脏.它定义了呼叫是如何流进和流出系统的.拨号计划用一种脚本语言写成的,Asterisk依照其中的指令响应外部触发.和传统电话系统相比,Asterisk的拨号计划是 ...

3. Spark The Definitive Guide Spark权威指南中文笔记

   目前在做Spark开发,所用到的参考资料便是Spark: The Definitive Guide.目前暂时没有中文版,为了记录学习和工作的过程,顺便等待中文版的推出,便将每章节的学习过程总结下来,以 ...

4. 《Spark权威指南》的翻译综述

   本文源地址:<Spark权威指南>的翻译综述 前言 本系列文章将对<Spark - The Definitive Guide - Big data processing made s ...

5. JavaScript权威指南 第11章JavaScript标准库

   JavaScript权威指南 第11章JavaScript标准库 第11章 JavaScript标准库 11.1 集合与映射 11.1.1 Set类 11.1.2 Map类 11.1.3 WeakMa ...

6. JavaScript权威指南 第15章 网络编程 第三部分

   JavaScript权威指南 第15章 网络编程 第三部分 可伸缩矢量图形 15.7.1 在HTML中使用SVG 15.7.2 编程操作SVG 15.7.3 通过JavaScript创建SVG图片 1 ...

7. 《Hadoop权威指南》第二章 关于MapReduce

   <Hadoop权威指南>第二章 关于MapReduce 目录 使用Hadoop来数据分析 横向扩展 注:<Hadoop权威指南>重点学习摘要笔记 1. 使用Hadoop来数据分 ...

8. HTML5 权威指南第 10 章 文档分节 学习笔记

   HTML5 权威指南第 10 章 文档分节 学习笔记 第 8 章 标记文字 内容从从文字出发,专注如何将单体内容正确的呈现出来:第 9 章 组织内容 内容从段落出发,专注如何将单体内容合理的放在段落中 ...

9. 《果壳中的C# C# 5.0 权威指南》 (09-26章) - 学习笔记

   <果壳中的C# C# 5.0 权威指南> ========== ========== ========== [作者] (美) Joseph Albahari (美) Ben Albahar ...

最新文章

1. 新方案-eclipse配置tomcat中文乱码另一种解决方案
2. Mac OS 怎么设置host
3. 手把手教你感情企划书
4. 更改sql-2008sa密码
5. python selenium翻页_Selenium翻页的实现方法实例
6. Css中部分知识点整理【笔记整理】
7. 【转】GIS原理学习
8. ascii码与hex转换c语言,ASCII与HEX对照转换表(示例代码)
9. 国有资产管理处组织召开新版固定资产管理系统操作培训会
10. HTML怎么使表格居中显示
11. python ttk style_python – 关于使用ttk.Style()的问题？
12. java计算机毕业设计游泳馆信息管理系统源程序+mysql+系统+lw文档+远程调试
13. 2015年上半年数据库系统工程师考试上午真题
14. centos 设置为北京时间
15. 协昌电子递交注册：年营收5.25亿 实控人顾挺曾任民警
16. oracle 中sql实现进一、去尾和四舍五入。
17. word常见问题_2 添加不同的页码
18. Javafx+MySQL 学生成绩管理系统
19. k8s 部署pod到指定node
20. 设置监察哨的顺序查找

热门文章

1. Oracle/Kingbase中用序列和触发器实现ID自增
2. 【CentOS7】selinux介绍及案例详解
3. 人工智能判断感冒与过敏性鼻炎问题
4. Kafka的零拷贝技术
5. css取第一个孩子,CSS：第n个孩子（）：后(css :nth-child() :after)
6. 小程序wx.showToast一闪，持续时间短
7. 选手得分规则为去掉一个最高分和一个最低分，然后计算平均得分，请编程输出某选手的得分。
8. ntp服务器地址是什么协议,ntp服务器地址的介绍与解释
9. 什么是WFS服务，发布WFS服务有什么用
10. 【其他】正则匹配淘口令