来源 | 「Stream Processing with Apache Flink」

作者 | Fabian Hueske and Vasiliki Kalavri

翻译 | 吴邪 大数据4年从业经验，目前就职于广州一家互联网公司，负责大数据基础平台自研、离线计算&实时计算研究

校对 | gongyouliu

编辑 | auroral-L

传统的数据基础架构

在过去的几十年，数据以及数据处理在企业和商业应用显得无处不在，随着时间的推进，多年以来，数据的收集和使用一直保持着持续增长的趋势，数据如何有效地管理成为每个公司的大事，为了更好的管理源源不断产生的用户数据，很多公司着力于设计和构建数据基础架构，通常情况下主要分为两种类型的架构：一是事务型数据处理，二是分析型数据处理，基于以上两种类型，下面我们讨论这两种类型的基础架构是如何管理和处理数据的。

事务型数据处理架构

很多公司在日常的业务场景中使用了五花八门的应用程序，比如说企业资源管理系统（ERP），客户关系管理系统（CRM）以及基于Web端的系统。这些系统通常在设计的时候会区分不同的数据处理层（即应用程序本身）和数据存储（事务型关系数据库），每个系统自成一套流程，如下图1-1 所示：

数据分析型处理架构

对于一个公司来说，存储在各种事务型数据库系统中的数据，通常能为公司的运营 决策提供有价值的参考依据，比如说，通过对订单处理系统的数据进行分析，可以掌握商品在一段时间内销售的增长率，以此来确定商品延期出货的根源，也可以用来预测未来的销售趋势，及时调整商品库存。然而，事务型数据通常分布在几个互不相连的数据库系统中，在需要进行联合分析时才显得更有价值，此外，不同数据库系统的数据在分析时通常需要转化为通用的处理格式。

与直接在事务型数据库执行分析查询有所不同，在数据分析型处理架构中，数据通常会被统一复制到数据仓库中，即专门用于数据分析查询的数据存储仓库，为了填充数据仓库，需要将事务型数据库系统管理的数据库的数据全部复制到数据仓库中，这个迁移数据的过程我们通常称为“提取-转换-加载”，也就是我们常说的ETL。ETL流程负责从事务型数据库（OLTP）抽取数据，根据一定的规范对数据进行验证、编码、去重以及数据结构等一系列操作进行转换，最终把处理好的数据加载到分析型数据库（OLAP）中，当ETL处理过程非常复杂的时候，这时候就需要考虑采用高性能的技术解决方案来满足需求，ETL通常被设置为一个定期运行的任务，目的是为了及时将事务型数据仓库的数据同步到数据仓库中，尽可能保证数据的完整性。

数据一旦被导入到数据仓库中，就可以用来做查询和分析，常见的有两类查询，第一种类型是定期报表查询统计，用于计算与业务相关的统计数据，如收入、用户增长或生产输出。把这些指标组合汇总到报告中，可以帮助管理层评估业务的总体健康状况。第二种类型是即席查询，旨在支撑特定问题的答案用来作为企业关键性决策的依据，例如，通过查询公司营收和投放广告支出，以评估营销活动方案的有效性，本质上，这两类查询说到底都是通过在数据仓库中执行批处理任务，从而得到计算结果，如图FIgure1-3所示：

现如今，Apache Hadoop生态系统的组件已经成为很多企业不可或缺的IT基础架构，而不再是将所有数据都插入到关系数据库系统中，而是将大量数据(如日志文件、社交媒体或web点击日志)写入Hadoop的分布式文件系统(HDFS)、S3或其他大容量数据存储系统，如Apache HBase，这些数据存储系统花费很小的成本就可以提供巨大的存储容量。存储系统中的数据可以通过SQL-on-Hadoop引擎进行查询和处理，例如Apache Hive、Apache Drill或Apache Impala。不可否认的是，这些基础架构设施基本上仍然与传统的数据仓库体系结构相同。

有状态流处理

实际上，所有数据的产生都可以看做连续的事件流，试想一下，用户与网站或手机APP应用产生的互动，订单的信息，服务器产生的日志或者传感器测量等等，统统都可以算是事件流。事实上，很难找到一个一次性生成有限并且完整的数据集的例子。有状态流处理是用于处理无界事件流的应用程序设计模式，适用于公司IT基础结构中的许多不同用例，在我们讨论这些用例之前，我们先简单解释一下有状态数据流处理的工作原理。

任何处理事件流的应用程序都应该是有状态的、能够被存储并且支持中间数据访问，而不仅仅是简单做实时数据记录的转换，当一个应用程序接收到一个事件时，能够通过从事件状态中读取或写入的数据进行任意的计算，对于状态来说，其本身可以存储并访问不同的介质，包括程序变量、内存、本地文件、嵌入式数据库或者外部数据库系统等。

Apache Flink通常将应用程序状态存储在本地内存中，或者嵌入式数据库中，比如Redis、RocksDB，由于Flink是一个分布式系统，因此需要保障本地状态的安全性，避免当应用程序失败或者机器故障引起数据丢失的情况发生。为了防止这种情况的出现，Flink通过定期对应用程序的状态做一致性checkpoint（检查点），类似于快照，并持久化到远程的数据库中，在下一章中，我们会对状态以及状态一致性和Flink checkpoint机制进行详细的讨论，本章节不做展开，图1-4展示的是一个有状态的Flink流式处理程序。

有状态流处理程序可以接收来自很多不同渠道和形式的数据，通常从事件流的日志提取事件注入流程序中进行计算，将事件日志存储并分发到事件流中，在这个过程中，事件会以追加的形式有序地持久化到日志中，这是一个有序的过程，一旦事件写入就无法修改顺序。写入事件日志的流可以被相同或不同的消费者多次读取，基于日志只能被追加的属性，所有的事件始终以完全相同的顺序发布给下游的消费者。在实际的使用中，有几个基于事件日志的开源软件可以作为我们的选择，比如Kafka、RabbitMQ、ActiveMQ，其中以Apache Kafka最受欢迎，Kafka作为时下最火热的消息中间件，可以集成到很多不同场景的系统架构中，受到很多云计算服务厂商的青睐。

出于很多不同的原因，将Flink上的有状态流应用程序和事件日志系统搭配使用是非常合适的，在这种体系结构中，事件日志可以用来持久化不断输入的事件，并可以按确定性顺序进行重放。在出现故障的情况下，Flink可以通过先前保存的检查点恢复状态并且重置事件日志上的读取位置来恢复有状态的流应用程序，然后流应用程序会根据检查点重放（快速转发）来自事件日志的输入事件，在流中重新进行有效回放，这项技术用于程序故障恢复，同时也可以用于更新应用程序、修复缺陷以及修复先前得出的结果、支持将应用程序迁移到不同的集群中运行或者用于区分程序版本进行A/B测试。

正如上面所说，有状态的流处理是一种通用且灵活性高的架构设计，可以用来处理不同应用场景下的用例，接下来，我们会介绍三种比较有代表性的应用：

2. 事件驱动型应用
3. 数据管道型应用
4. 数据分析型应用

事件驱动型应用

1. 实时推荐（比如用户在浏览电商网站时推荐产品）
2. 模式检查或复杂事件处理（例如用于信用卡交易中的欺诈检测）
3. 异常检测（比如试图侵入计算机网络检测）

图1-5中的应用程序通过事件日志进行关联， 一个应用程序将其输出发送到下游事件日志，上游程序的输出结果可以作为输入事件给另一个应用程序进行消费，事件日志可以将发送器和接收器之间的关联关系实现解耦，并提供异步、非阻塞事件传输。每个应用程序都可以是有状态的，并且可以在本地管理自己的状态，而不需要访问外部数据存储，不仅如此，每个 应用程序也可以单独或者关联起来运行。

相对于传统的事务型应用和微服务来说，事件型驱动应用有几个比较明显的优势，事件驱动型应用无须查询远程数据库，本地数据访问使得它具有更高的吞吐和更低的延迟。而由于定期向远程持久化存储的 checkpoint 工作可以异步、增量式完成，因此对于正常事件处理的影响甚微。事件驱动型应用的优势不仅限于本地数据访问。传统分层架构下，通常多个应用会共享同一个数据库，因而任何对数据库自身的更改（例如：由应用更新或服务扩容导致数据布局发生改变）都需要谨慎协调。反观事件驱动型应用，由于只需考虑自身数据，因此在更改数据表示或服务扩容时所需的协调工作将大大减少。

另外，事件驱动型应用程序对运行它们的流处理器有相当高的要求，不是所有的流处理器都适合运行事件驱动型应用程序。事件驱动型应用会受制于底层流处理系统对时间和状态的把控能力，Flink 诸多优秀特质都是围绕这些方面来设计的。它提供了一系列丰富的状态操作原语，允许以精确一次的一致性语义合并海量规模（TB 级别）的状态数据。此外，Flink 还支持事件时间和丰富的窗口逻辑操作，而且它内置的 ProcessFunction 支持细粒度时间控制，方便实现一些高级业务逻辑。同时，Flink 还拥有一个复杂事件处理（CEP）类库，可以用来匹配数据流中的模式，Apache Flink符合以上所有的要求，是事件驱动型应用非常好的选择。Flink 中针对事件驱动应用有一个天然的特性—— savepoint。savepoint 是一个一致性的状态快照，它可以用来初始化任意状态兼容的应用。在完成一次 savepoint 后，即可放心对应用进行升级或扩容，还可以启动多个版本的应用来执行 A/B 测试。

数据管道型应用

数据分析型应用

数据分析型应用根据实时性大致可以分为批处理和流处理两种类型。

批处理分析

流式处理分析

与批处理分析型应用不同的是，流式处理不需要周期性去触发作业，而是实时地处理不断输入的事件流，通过及时计算并合并最新的结果，达到低延迟的效果，类似数据库用于更新物化视图技术。通常来说，流应用程序将其计算结果存储在支持高效更新的外部数据存储中，例如数据库或K-V类型的存储系统中，另外，流分析应用的实时更新结果可应用到仪表盘，如图1-6所示。

流分析应用程序除了能用更短的时间合并事件的分析结果之外，还有一个相对不太明显的优势，传统的数据分析管道由几个独立的组件组成，比如ETL作业、存储系统，以及基于hadoop生态的数据处理引擎和调度器触发作业或查询，通常需要将这些组件进行编排，当程序出现故障时需要花费比较多的时间去排查错误。与之相反的是，有状态流应用程序的流处理器会负责处理所有这些数据处理步骤，包括事件摄入、连续计算（中间状态维护）和结果更新。不但如此，Flink提供exactly-once（状态一致性）机制，可以保证程序出现异常时进行正确的恢复并调整集群计算资源，诸如Flink这样的流计算引擎拥有事件时间机制、高吞吐量的特性，可以很短的时间内处理海量数据，输出正确的结果。

流分析型应用有下面几种较典型的场景：

开源流处理的演进

1. 开源意味着免费和开放，门槛低，容易被大众接受和使用。
2. 由于开源社区的努力，很多开放者贡献了自己的代码，使得开源技术往更好、易用、高性能等好的方向快速发展，促进开源技术变得越来越成熟。

历史回顾

Flink 速览

Apache Flink就是第三代流式处理引擎中的典型代表，具备很多卓越的特性，如低延迟、高吞吐量等，在这里列举Flink的部分优势：

1. 丰富的时间语义，支持三种时间语义，processing-time、ingest-time、event-time，其中event-time语义提供了一致性结算结果支持，可以处理乱序数据，而process-time语义适用于实时性要求非常严格的应用程序。
2. 状态一致性保证。
3. Flink实现了毫秒级别延迟，并且能够每秒处理数百万个事件，Flink应用程序可以扩展到在数千个cpu core上运行。
4. Flink具有层次分明的API，提供了三种不同的 API，每一种 API 在简洁性和表达力上有着不同的侧重，并且针对不同的应用场景。本书涵盖了DataStream API和底层ProcessFunction，常用于流处理操作(如窗口处理和异步通信)，还提供了精确控制状态和时间的接口。包括了SQL&Table API，本书不做展开说明。
5. Flink提供了丰富的connector连接器与外部存储系统连接，如Apache Kafka、Apache Cassandra、Elasticsearch、JDBC、Kinesis、HDFS和S3等存储媒介。
6. 保证7x24小时全天候服务，提供高可用方案并将程序托管在YARN、Mesos和Kubernetes等高效的资源管理器上，动态调控资源，提升资源利用率。
7. 能够更新应用程序代码并将作业迁移到不同的Flink集群中，而不会丢失应用程序的状态。
8. 细粒度监控集群各项指标，以提前做好预警处理工作。
9. 支持流批一体化。

除了以上的特性，Flink封装了很多易用的API接口，这对开发人员来说是一个非常友好的框架，在开发和测试的过程中，还可以在单核JVM处理器中通过IDE工具进行调试。

运行首个Flink 应用

接下来，我们指导你在本地集群运行你的第一个Flink应用程序，对随机生成的温度传感器数据进行转换和聚合操作，让你对Flink应用有个大概的了解。首先准备Flink集群运行的环境，JDK 1.8，Unix或Centos 系统，实在不行也可以在window是系统上安装虚拟机环境。

tar xvfz flink-1.7.1-bin-scala_2.12.tgz 

3.启动集群
$ cd flink-1.7.1$ ./bin/start-cluster.sh
Starting cluster.Starting standalonesession daemon on host xxx.Starting taskexecutor daemon on host xxx.

4.在浏览器输入 http:// localhost:8081，进入flink webUI页面，默认只有一个slot，如图1-8所示。

5.下载本书示例的JAR文件

$ wget https://streaming-with-flink.github.io/\examples/download/examples-scala.jar

6.在本地集群运行样例程序

$ ./bin/flink run \-cio.github.streamingwithflink.chapter1.AverageSensorReadings \examples-scala.jar
提供任务之后会出现以下提示信息：Starting execution of programJob has been submitted with JobIDcfde9dbe315ce162444c475a08cf93d9

7.在Flink webUI点击Running Job可以看到提交运行的任务，点击对应的任务可以看到更多详细的指标，如图1-9所示。

8.计算结果会被标准输出到默认的文件中，可以在 安装目录的log文件夹下看到，执行下面的指令即可。

$ tail -f ./log/flink-<user>-taskexecutor-<n>-<hostname>.out 执行命令之后，就会看到下面的输出信息，包括了SensorReading对象的id，时间戳，和平均温度SensorReading(sensor_1,1547718199000,35.80018327300259)SensorReading(sensor_6,1547718199000,15.402984393403084)SensorReading(sensor_7,1547718199000,6.720945201171228)SensorReading(sensor_10,1547718199000,38.101067604893444)

9.一旦提交任务，程序就会一直运行直到任务完成，不管最后是执行成功还是执行失败，也可以通过命令或者在界面上点击cancel按钮取消任务。

10.关闭集群

$ ./bin/stop-cluster.sh

以上我们完成了Apache Flink本地集群的安装部署，并且试着运行了第一个流应用程序，当然啦，目前为止，我们只是简单认识了Flink，对于Flink来说，可能勉强算得上刚入门，关于Apache Flink这个框架还有非常多的内容需要我们不断去学习，这也是本书的价值所在。

总结