Apache Flink 学习实践

参考书籍

Apache Flink 官方文档

基于Apache Flink的流处理：流式应用基础、实现及操作 -(美)比安·霍斯克(Fabian Hueske),(美)瓦西里基·卡拉夫里(Vasiliki Kalavri) 著 崔星灿 译 中国电力出版社

基础概念

Apache Flink

  Apache Flink 是一个分布式流处理引擎，它提供了直观且极富表达力的 API 来实现有状态的流处理应用，并且支持在容错的前提下高效、大规模地运行此类应用。

ETL Extract-Transform-Load

  对于分析类查询，我们通常不会直接在事务型数据库上执行，而是将数据复制到一个专门用来处理分析类查询的数据仓库。为了填充数据仓库，需要将事务型数据库系统中的数据拷贝过去。这个向数据仓库拷贝数据的过程被称 为提取-转换-加载( Extract-Transform-Load , ETL)。
  ETL 的基本流程是 : 从事务型数据库中提取数据，将其转换为通用表示形式(可能包含数据验证、 数据归一化 、编码、去重、表模式转换等工作)，最终加载到分析型数据库中。为了保持数据仓库中的数据同步， ETL 过程需要周期性地执行。

有状态的流式应用

  处理事件流的应用，如果要支持跨多条记录的转换操作，都必须是有状态的，即能够存储和访问中间结果。应用中访问的状态有多种可选的存储位置，例如:程序变量、本地文件、嵌人式或外部数据库等。

  Apache Flink 会将应用状态存储在本地内存或嵌入式数据库中。 由于采用的是分布式架构，Flink 需要对本地状态予以保护，以避免因应用或机器故障 导致数据丢失。为了实现该特性，Flink 会定期将应用状态的 一致性检查点 (checkpoint)写入远程持久化存储。

   有状态的流处理应用通常会从事件日志中读取事件记录。事件日志负责存储事件流并将其分布式化。由于事件只能以追加的形式写入持久化日志中，所以其顺序无需在后期改变。写入事件日志的数据流可以被相同或不同的消费者重复读取。得益于日志的追加特性，无论向消费者发布几次，事件的顺序都能保持一致。

事件驱动型应用

   与事务型应用和微服务架构相比，事件驱动型应用有很多优势:

• 访问本地状态的性能要比读写远程数据存储系统更好;
• 伸缩和容错交由流处理引擎完成;
• 以事件日志作为应用的输入，不但完整可靠，而且还支持精准的数据重放。
• 此外Flink可以将 应用状态重置到之前的某个检查点，从而允许应用在不丢失状态的前提下更新或扩缩容。

安装运行

官方下载

1. Apache Flink 网站 下载官方发行版
2. 解析文件 tar xvfz flink-<version>.tgz
3. 启动本地Flink集群 flink-<version>/bin/start-cluster. sh
4. 打开本地 Flink Web UI，观察 TaskManager

运行作业

1. 下载作业DEMO wget https://streaming-with-flink .github.io/examples/download/examples-scala.jar
2. 在本地集群运行DEMO ./bin/flink run -c io.github.streamingwithflink.chapterl.AverageSensorReadings examples-scala.jar
3. 检查Flink Web UI，观察 Running Jobs 列表
4. 作业的输出默认在 flink-<version>/log/link-<user>-taskexecutor-<n>-<hostname>.out中，可以使用tail -f观察日志输出。

停止

1. 由于应用是流，它会一直运行下去，直到手动取消，可在Flink Web UI中手动选中作业，单击Cancel按钮
2. 停止Flink集群 flink-<version>/bin/stop-cluster. sh

DataFlow

基本概念

  DataFlow 程序通常表现为有向图，图中顶点称为算子，表示计算，是基本功能单元；边表示数据依赖关系；没有输入端的算子称为数据源，没有输出端的算子称为数据汇。算子从输入端获取数据进行计算后产生的数据发往输出端以供后续处理。

  上图被称作逻辑图，在实际使用分布式处理引擎时，每个算子可能会在不同的物理机器上运行多个并发任务。在逻辑图中顶点代表算子，在物理图中顶点代表任务。转换和计算算子都包含两个并行任务，每个任务负责计算一部分数据。

数据并行和任务并行

• 数据并行
  • 将数据分组，执行同一操作的多个任务并行执行在不同数据集上。
  • 将计算负载分配到多个节点上，从而允许更大规模的数据
• 任务并行
  • 不同算子下的任务(基于相同或不同的数据)并行进行计算。
  • 可以更好利用集群资源

数据交换策略

• 转发
  • 发送端任务与接收端任务一对一数据传输，如果两端在同一台物理机上，可以避免网络通信开销
• 广播
  • 将全部数据发往下游算子的全部并行任务，该策略会复制多份数据且涉及网络通信开销
• 基于键值
  • 根据某一属性对数据分区，保证相同的数据交由同一任务处理
• 随机
  • 将数据均匀分配给算子，实现任务之间的负载均衡

流处理

延迟与吞吐

• 延迟
  • 延迟表示从接收事件到输出处理效果所需的时间，流处理中以时间片为单位测量。
  • 根据应用不同，关注的延迟模式分为平均延迟、最大延迟、延迟的百位数值
    • 平均延迟：平均每条数据处理的时间
    • 延迟的百位数值：95%的延迟在10ms内
  • 保证低延迟对流式应用至关重要，如系统告警、诈骗识别
• 吞吐
  • 吞吐是衡量系统处理能力（处理速率）的指标，表示系统每单位时间所能处理的事件数量。
  • 事件到达速率过高会使系统被迫缓冲事件，如果系统持续以高速率接收事件，缓存区溢出将会使数据丢失，这种情形称为背压 (backpresssure)

  延迟与吞吐并非相互独立的指标，如果数据在处理中所花费时间太久（延迟太高），将难以保证高吞吐；同理，如果数据速率过高，将会积压事件，必须等待一段时间才能处理。

流上的操作

• 转换操作
  • 只经过一次算子的操作，事件被逐个读取，通过算子内置或用户自定义的函数转换成相应的输出。
  • 算子既可以接收多个流输出多个流，也可以分割单个流或合并多条流改变DataFlow的结构。
• 滚动聚合
  • 根据每个到来的事件持续更新结果
  • 滚动聚合是有状态的，通过新到来的事件合并到目前状态来生成更新后的聚合值。
• 窗口操作
  • 由于流是无限的，应用需要为特定维度下有限的数据集合产生计算结果，如每5分钟的路况、最近30分钟的交易数据等。窗口操作会持续创建称为桶的有限事件集合，通常会根据事件的时间或其他属性分配到不同桶中。
  • 窗口策略决定了什么时间创建桶、事件如何分配到桶中、桶内数据什么时间参与计算。
  • 常见的窗口类型
    • 滚动窗口：将事件分配到长度固定且互不重叠的桶中，当窗口边界通过后，将桶内所有事件发给计算函数处理。（例如固定时间或数量的窗口）
    • 滑动窗口：将事件分配到长度固定允许互相重叠的桶，意味着每个事件都有可能属于多个桶。
    • 会话窗口：在线分析用户行为应用中，需要将用户的一系列活动事件进行分桶，会话的长度并发事件预定好，而是和实际数据有关（例如活动的间隔）。

时间语义与一致性模型

  现实世界的系统、网络以及通信信道往往都充斥着缺陷，如何在这种情况下提供精准、确定的结果至关重要；此外流式应用需要长年累月的运行，必须保证其出错时状态能进行可靠的恢复。

  某个游戏公司的服务会分析用户玩在线玩手游时产生的事件。该应用将用户组织成不同团队，并会收集每个团队的活动信息，这样就能基于团队成员完成游戏目标的速度，提供游戏奖励(例如，团队所有成员在一分钟内消除了500个泡泡，他们就会提升一级)。爱丽丝是个铁杆玩家，每天早晨上班路上都会玩这个游戏。但是有个问题:爱丽丝每天乘地铁上班，而她的上班路上有几站地铁手机会断网，考虑如下情况:爱丽丝开始消泡泡的时候手机还能联网向服务器发送事件。突然，地铁开进隧道，手机断网了。爱丽丝继续玩她的，此时游戏产生的事件会缓存在手机里。在地铁离开隧道，爱丽丝重新上线后，之前缓存的事件才会发送给服务器。此时服务器该怎么办?

  在上述示例中一分钟的含义又是什么?需要把爱丽丝离线的时间考虑在内吗?

时间语义

• 处理时间
  • 指流计算算子所在机器的本地时钟时间，基于处理时间的窗口会包含恰好在一段时间内达到窗口算子的事件。
  • 在爱丽丝游戏的例子中，处理时间窗口在她处于离线时会继续进行计时，因此不会计算她离线的那些事件。
• 事件时间
  • 事件时间是数据流中事件发生的实际时间，它以附加在事件流中的时间戳为依据。这些时间戳通常在事件进入数据管道之前就存在，事件时间能准确将事件分配到窗口中，反映出真实发生的情况。基于事件时间的操作是可预测的，其结果具有准确性，无论数据流的处理速率如何，数据到达的顺序怎样，基于事件时间的窗口都会生成 同样的结果 。
  • 使用事件时间的挑战在于如何处理延时事件，普遍存在的乱序问题也可以由此解决；假设有另一位叫鲍勃的用户与爱丽丝同时在玩这个游戏，但由于鲍勃的网络提供商依然能稳定联网并向服务端发送游戏事件。依靠事件时间，我们能保证数据在乱序的情况下依然准确：结合可重放的数据流和时间戳带来的结果确定性，意味着通过重放数据来分析/计算历史数据，如同是实时产生的一样。此外，通过并行或扩容可以将计算快进到现在，作为实时应用继续运行。

水位线

  由于现实系统中的不确定性，我们无法得知某个时间点之前的事件已全部到达，如何决定事件时间窗口的触发策略以处理窗口里的数据呢?

  水位线是一个全局进度指标，表示一个时间点，其之后不会再有延迟事件到达。当算子接收到时间为T的水位线，就可以认为不会再有时间戳小于等于T的事件会到达了，可以触发窗口计算或排序操作了。

  水位线允许应用在结果的准确性与延迟之间做出权衡，激进的水位线策略保证了低延迟，但随之而来的是低可信度，延迟事件可以会在水位线之后到来，必须加一些额外处理逻辑应对；反之虽然得以保证可信度，但数据的实时性受到影响。

  在现实应用中，系统无法获取足够的信息来确定完美的水位线，以游戏举例，现实中根据无法得知用户会离线多久，他们可能在过隧道，登飞机，或是直接退坑了。直接依赖水位线并不是高枕无忧的，流处理系统很关键的一点是提供某些机制来应对那些迟到于水位线之后的事件，可能会增加额外逻辑来数据修正，可能会写入日志或者干脆丢弃。

流处理中的状态

• 状态
  • 由于流式算子处理的都是无穷无尽的数据，为了限制状态大小，算子通常只保留目前为止事件的摘要或概览，这种摘要可能是一个数值或是自定义的数据结构。
• 故障
  • 对于算子任务，一般都会执行以下步骤：接受事件并存储、选择性更新状态、产生输出。其任一步骤都有可能产生故障，流处理系统通过不同的保障策略来定义故障时的行为。
• 保障策略
  • 至多一次 可以说是没有保障，既不恢复丢失的状态，也不重放丢失的事件，可以保证低延迟。
  • 至少一次 从源头或缓冲区重放事件，一般事件日志会写入永久存储，在故障时可以重放。另一种方式是采集记录确认，当所有任务确认某个事件都处理完毕后再丢弃事件。
  • 精确一次 最严格，也最难实现，表示每个事件都没有丢失，而且对于内部状态的更新只有一次，在至少一次保障的前提上，保证内部状态的一致性。

Apache Flink 架构

  Flink只专注于流处理任务，对于分布式系统的集群资源分配与管理依赖于集成的其他基础设施，如YARN、Mesos、Kubernetes等；也并没有提供分布式持久化存储，而是利用了其他文件系统如HDFS、S3等；它依赖Zookeeper来完成高可用中的Leader选举工作。

Flink组件

  Flink有四个不同组件：JobManager、ResourceManager、TaskManager、Dispatcher，Flink本身是由Java和Scala编写的，因此所有组件都在JVM上运行。

组件协同

任务执行

  左侧逻辑图中应用包含5个算子，A 与 C 是数据源，E 是数据汇，C 与 E 并行度是2，其它为4。由于最大并行度是4，所以需要4个处理槽。如果每个TaskManager的处理槽是2，那么运行2个TaskManager可以满足需求。

  并行度为4的算子会被分配到四个槽中，C 与 E 算子各被分配到两个槽中：

• C：Slot 1.1、Slot 2.1
• E：Slot 1.2、Slot 2.2

  TaskManager会在同一个JVM进程内，其优缺点如下：

• 其中的多个任务可以在不经过网络的情况下进行高效数据交换，然而任务过于集中会使TaskManager负载变高，导致性能下降，需要确认平衡性。
• 以多线程的方式执行任务，线程相比进程更轻量，通信开销更低，但无法隔离彼此，只要有一个任务异常，有可能会影响到其他任务，因此需要根据实际情况进行调优。性能和资源隔离也是需要取舍之一。

Flink 高可用性

TaskManager高可用

  为了执行全部任务，Flink需要足够的处理槽，当TaskManager出现故障导致处理槽不足时，JobManager会向ResourceManager申请更多的处理槽，直至处理槽可用并完成任务重启。

JobManager高可用

  JobManager基于Zookeeper实现高可用，JobManager会将任务的JobGraph以及任务元数据（例如应用的jar包、检查点）写入至远程存储，并将路径写入至Zookeeper，因此所有用于JobManager重启恢复的数据都在远程存储上，而Zookeeper有这些数据的存储路径。

如果Flink是在容器环境中以库模式部署，编排器会自动重启JobManager

Flink 数据传输

   Flink中的TaskManager负责将数据由发送端传输到接收端，它的网络模块将数据放到缓冲区以批次发送，此机制目的是高效利用网络带宽，实现高吞吐。同个TaskManager内部的数据交换不经过网络，不同的TaskManager之前的数据交换需要经过网络，但同一对TaskManager的网络连接共享给内部的任务。接收端通过自己的缓冲区与各发送端的数据积压来给发送端数据额度，发送端根据数据额度发送对应的数据量。

Flink 事件时间处理

• 时间戳
  • 事件时间模式下，Flink应用所处理的数据有一个内部8字节的unix时间戳，它以元数据形式附加在事件记录上。
• 水位线
  • Flink利用一些包含unix时间戳的特殊事件记录来实现水位线。水位线之后的事件记录被称为迟到事件，支持以不同的策略来处理。
  • Flink采用以下机制确保水位线单调递增，针对水位线之后迟到的事件使用其他策略处理。
    • 算子有多个输入流时，事件时间时钟会受限于最慢的输入流。
    • 如果任一输入流水位线没有前进，任务的事件时钟就不会更新，导致算子的处理会一直延迟。

Flink 状态管理

   Flink中，根据作用域的不同，状态可以分为两类：算子状态(operator state) 和 按键分区状态(keyed state)。

• 算子状态 operator state
  • 在算子实例被创建时生成，并在算子实例被销毁时销毁。
  • 每个算子实例都有自己的状态副本，实例在处理时可以访问和修改。
  • 多个算子实例之间算此独立，互不干扰
  • 算子状态有三类原语
    • 列表状态 List State 状态条目列表
    • 联合列表状态 Union List State 列表状态用于存储与特定键相关联的列表数据，而联合列表状态用于在多个键之间共享列表数据
    • 广播状态 Broadcast State 保证算子的每个任务状态都相同

   算子状态扩缩容

• 按键分区状态 keyed state
  • Flink 会按照算子输入记录所定义的键进行维护或访问，Flink 为每个键值都维护了一个状态实例，该实例位于处理那个键值对应的记录的算子实例任务上。
  • Flink 通过将相同键的数据分配到相同的任务中，保证了按键分区状态（Keyed State）的正确分配和访问。
  • 扩缩容时将所有键值重新分组，

   按键分区状态扩缩容

Flink 检查点 CheckPoint、保存点 SavePoint、与状态恢复

简单应用的一致性检查点 与检查点恢复

• 单个算子虽然会重复处理事件（如图中序列为4的事件），但由于算子状态会重置到处理事件前的状态，因此能实现精准一次的状态一致性。
• Flink会周期性的生成一致性检查点，一旦发生故障，Flink 会利用最新一次的检查点恢复应用状态。

检查点算法

   Flink中的检查点算法使用一类称为检查点屏障 checkpoint barrier的特殊记录（和水位线相似），它将流分成了两个部分：之前的数据已经完成检查点，而之后的数据则属于下一个检查点，每个检查点屏障都带有一个编号。

• 数据源对检查点的处理

• 任务对检查点的处理

• sink对检查点的处理

保存点

• 检查点会周期性的生成，并根据配置的策略自动丢弃，应用停止后，检查点也随之删除。
• 原则上保存点与检查点的生成算法一样，可以把保存点看做包含一些额外元数据的检查点，保存点由用户显式触发生成并且不会被Flink自动删除。

Apache Flink 实践

• 创建DataStream API 项目
• 使用Table API