流／批／OLAP 一体的 Flink 引擎介绍｜ 青训营笔记
# 流／批／OLAP 一体的 Flink 引擎介绍｜ 青训营笔记
这是我参与「第四届青训营 」笔记创作活动的的第 2 天
# Apache Flink 概述
Apache Flink 是一个框架和分布式处理引擎，用于对无界和有界数据流进行状态计算。
# Apache Flink 诞生背景
什么是大数据：指无法在一定时间内用常规软件工具对其进行获取、存储、管理和处理的数据集合
- 海量化：海量数据
- 多样化：数据源和数据种类多样，如非结构化数据、半结构化数据和结构化数据
- 快速化：数据产生和处理速度非常快
- 价值化：数据价值密度低、整体价格密度高
大数据计算架构发展历史
史前阶段～2006 ：
- 传统数仓
- Oracle
- 单台大型机器处理
- 黑箱使用 （Google “三驾马车”）
- GFS 分布式文件存储
- MapReduce 分布式计算框架
- BigTable 分布式结构化数据存储系统
Hadoop：
- 分布式
- Map-Reduce（双算子 API）
- 离线计算
Spark：
- 批处理
- 流处理
- SQL 高阶 API（更多语义、使用简单）
- 内存迭代计算
Flink：
- 流计算
- 实时、速度快
- 流批一体
- Streaming/Batch SQL
# 为什么需要流计算？
- 大数据实时性带来价值更大：监控场景、金融风控、实时推荐等
- 大数据实时性要求，带来计算架构模式的变化，从批式到流式
- 批式计算
- 离线计算，非实时
- 静态数据集
- 小时 / 天等周期性计算
- 流式计算
- 实时计算、快速、低延迟
- 无限流、动态、无边界数据源
- 7*24 h 不断运行
- 流批一体
- 批式计算
传统数据处理架构
事务处理
分析处理
将数据从业务数据库复制到数仓，再进行分析和查询
有状态的流式处理
# 为什么 Flink 会脱颖而出
流计算引擎的对比
| Storm | Spark Streaming | Flink | |
|---|---|---|---|
| Streaming Model | Native | mini-batch | Native |
| 一致性保证 | At Least/Most Once | Exactly-Once | Exactly-Once |
| 延迟 | 低延迟（毫秒级） | 高延迟（秒级） | 低延迟 （毫秒级） |
| 吞吐 | 低 | 高 | 高 |
| 容错 | ACK | RDD Based Checkpoint | Checkpoint（Chandy-Lamport） |
| StateFul | No | Yes（DStream） | Yes（Operator） |
| SQL 支持 | No | Yes | Yes |
流式计算引擎发展历史
- Storm：History of Apache Storm and lessons learned - thoughts from the red planet (opens new window)；
- Storm API 的 low-level 以及开发效率低下；
- 一致性问题：Storm 更多考虑到实时流计算的处理时延而非数据的一致性保证；
- Spark Streaming：An Architecture for Fast and General Data Processing on Large Clusters (opens new window)；
- Spark Streaming 相比于 Storm 的低阶 API 以及无法正确性语义保证，Spark 是流处理的分水岭：第一个广泛使用的大规模流处理引擎，既提供较为高阶的 API 抽象，同时提供流式处理正确性保证。
- Flink：从产品技术来看，Flink 作为一个最新的实时计算引擎，具备如下流计算技术特征：
- 完全一次保证：故障后应正确恢复有状态运算符中的状态；
- 低延迟：越低越好。许多应用程序需要亚秒级延迟；
- 高吞吐量：随着数据速率的增长，通过管道推送大量数据至关重要；
- 强大的计算模型：框架应该提供一种编程模型，该模型不限制用户并允许各种各样的应用程序在没有故障的情况下，容错机制的开销很低；
- 流量控制：来自慢速算子的反压应该由系统和数据源自然吸收，以避免因消费者缓慢而导致崩溃或降低性能；
- 乱序数据的支持：支持由于其他原因导致的数据乱序达到、延迟到达后，计算出正确的结果；
- 完备的流式语义：支持窗口等现代流式处理语义抽象；
- Google Dataflow Model 的开源引擎实现。
为何选择 Flink
- 精确一次（Exactly-Once）的状态一致性保证
- 支持事件时间（event-time）和处理时间（processing-time）语义
- Stateful 状态容错 (Checkpoint)
- Dataflow 编程模型 Window 等高阶需求支持友好
- 流批一体
# Apache Flink 开源生态
- 流批一体：支持流计算和批计算
- OLAP：Fink 支持 OLAP 短查询场景
- Flink ML： pyFlink、Alink、AIFlow 等生态支持 Flink 在 ML 场景的应用
- Gelly：图计算
- Stateful Function：支持有状态的 FAAS 场景
# 流批一体的 Apache Flink 架构
# Flink 整体架构
作业管理器（JobManager）：负责整个任务的协调工作，包括：调度 task、触发协调 Task 做 Checkpoint、协调容错恢复等，核心有下面三个组件：
- 资源管理器（ResourceManager）：负责 slot 资源的管理和调度，Task manager 拉起之后会向 RM 注册
- 分发器（Dispatcher）: 接收作业，拉起 JobManager 来执行作业，并在 JobMaster 挂掉之后恢复作业
- JobMaster: 管理一个 job 的整个生命周期，会向 ResourceManager 申请 slot，并将 task 调度到对应 TM 上。它还运行 Flink WebUI 用来提供作业执行信息。
任务管理器（TaskManager）：负责执行一个 DataFlow Graph 的各个 task 以及 data streams 的 buffer 和数据交换。
# Flink 分层架构
SDK 层：Flink's APIs Overview (opens new window)；目前主要有三类，SQL/Table、DateStream、Python
执行引擎层（Runtime 层）：执行引擎层提供了统一的 DAG，用来描述数据处理的 Pipeline，不管是流还是批，都会转化为 DAG 图，调度层再把 DAG 转化成分布式环境下的 Task，Task 之间通过 Shuffle 传输数据；
- 调度：Jobs and Scheduling (opens new window)；
- Task 生命周期：Task Lifecycle (opens new window)；
- Flink Failover 机制：Task Failure Recovery (opens new window)；
- Flink 反压概念及监控：Monitoring Back Pressure (opens new window)；
- Flink HA 机制：Flink HA Overview (opens new window)；
状态存储层：负责存储算子的状态信息
资源调度层：目前 Flink 可以支持部署在多种环境
# Flink 作业示例
Flink 作业示例（Flink Learn: Hands-On Training (opens new window)）
DataFlow（程序与数据流）：所有的 Flink 程序都是由三部分组成的： Source 、Transformation 和 Sink。Source 负责读取数据源，Transformation 利用各种算子进行处理加工，Sink 负责输出。
ExecutionGraph（执行图） 生成：StreamGraph （DataStream API Code） --> JobGraph --> ExecutionGraph（Parallelized）--> 物理执行图。
StreamGraph：是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。
JobGraph：StreamGraph 经过优化后生成了 JobGraph，提交给 JobManager 的数据结构。主要的优化为，将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化 / 反序列化 / 传输消耗。
ExecutionGraph：JobManager 根据 JobGraph 生成 ExecutionGraph。ExecutionGraph 是 JobGraph 的并行化版本，是调度层最核心的数据结构。
物理执行图：JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个 TaskManager 上部署 Task 后形成的 “图”，并不是一个具体的数据结构。
Parallelism（并行度）：Flink 程序的执行具有并行、分布式的特性。在执行过程中，一个流（stream）包含一个或多个分区（stream partition），而每一个算子（operator）可以包含一个或多个子任务（operator subtask）。一个特定算子的子任务（subtask）的个数被称之为其并行度（parallelism）。
假设作业的 sink 算子的并发配置为 1，其余算子并发为 2。
紧接着会将上面的 Streaming DataFlow Graph 转化 Parallel Dataflow (内部叫 Execution Graph) :
OperatorChain（任务链）：相同并行度的 one to one 操作，为了更高效地分布式执行，Flink 会尽可能地将不同的 operator (chain）在一起形成 Task。这样每个 Task 可以在一个线程中执行，内部叫做 OperatorChain。将算子链接成 task 是非常有效的优化：它能减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。如下图的 Key Agg 和 Sink 算子可以 Chain 在一起。
下面以 Kafka 为数据源 在流中作 WordCount 示例 并存储到 Redis
def main(args: Array[String]): Unit = {
// 创建流处理器
val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
// 设置并行度
environment.setParallelism(1)
// kafka配置
val properties = new Properties()
properties.put("bootstrap.servers", "master:9092")
// 构建kafka消费者source
val kafka = new FlinkKafkaConsumer[String]("order", new SimpleStringSchema(), properties)
// 构建一个redis配置池
val poolConfig: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder().setHost("master").setPort(6379).build()
// 在流中添加kafka source
val value: DataStream[String] = environment.addSource(kafka)
// 映射
val result: DataStream[(String, Long)] = value.map(f => {
(f, 1L)
})
// 根据key分区
.keyBy(_._1)
// 时间窗口 5分钟滑动 统计一次单词出现次数
.timeWindow(Time.minutes(5))
// 输入类型 变化类型 输出类型
.aggregate(new AggregateFunction[(String, Long), Long, Long] {
override def createAccumulator(): Long = 0L
override def add(in: (String, Long), acc: Long): Long = acc + 1
override def getResult(acc: Long): Long = acc
override def merge(acc: Long, acc1: Long): Long = acc + acc1
})
// 在流中添加redis sink
result.addSink(new RedisSink[(String, Long)](poolConfig,
// 重写RedisMapper中的方法
new RedisMapper[(String, Long)] {
override def getCommandDescription: RedisCommandDescription = new RedisCommandDescription(RedisCommand.SET)
override def getKeyFromData(t: (String, Long)): String = t._1
override def getValueFromData(t: (String, Long)): String = t._2.toString
}))
environment.execute("job submit")
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# 为何需要流批一体
- 一些业务场景，除了实时的数据统计需求，为了确认运营或产品的效果，用户同时还需要和历史数据做比较，比如，抖音一些直播数据的统计；
- 按照时间段统计数据信息需要 Flink 的批式处理
- 数据源（埋点日志、业务消息、数据库）-> 实时数仓 Flink（Kafka 基于流式数据进行统计）OR 离线数仓 Hive/Spark
- 人力成本高：批、流两套系统，相同逻辑需要开发两遍；
- 数据链路冗余：本身计算内容是一致的，由于是两套链路，相同逻辑需要运行两遍，产生一定的资源浪费；
- 数据口径不一致：两套系统、两套算子、两套 UDF，通常会产生不同程度的误差，这些误差会给业务方带来非常大的困扰。
# 流批一体的挑战
| 流式计算 | 批式计算 | |
|---|---|---|
| 计算方式 | 实时计算 | 离线计算 |
| 迟延迟要求 | 延在秒级以内 | 处理时间为分钟到小时级别，甚至天级别 |
| 任务时长 | 0-1s | 10s-1h+ |
| 数据流 | 数据流为无限数据集 | 有限数据集 |
| 时延 | 低延迟、业务会感知运行中的情况 | 实时性要求不高，只关注最终结果产出时间 |
| 场景 | 广告推荐、金融风控 | 搜索引擎构建索引、批式数据分析 |
# Flink 如何做到流批一体
- 批式计算是流式计算的特例，Everything is Streams，有界数据集（批式数据）也是一种数据流、一种特殊的数据流； 无边界数据集是一种数据流，一个无边界的数据流可以按时间切段成一个个有边界的数据集，所以有界数据集（批式数据）也是一种数据流。 因此可以用一套引擎解决两种场景，但需要对不同场景支持不同扩展性、实现不同优化策略；一个无边界数据集也是数据流，可以切割成一段段有限数据集，因此可以使用同一套 API。
- SQL 层：支持有无边界的输入
- DataStream API 层统一，批和流都可以使用 DataStream API 来开发；
- Scheduler 层架构统一，支持流批场景；
- Failover Recovery 层 架构统一，支持流批场景；
- Shuffle Service 层架构统一，流批场景选择不同的 Shuffle Service；
# 流批一体的 Scheduler 层
Scheduler 主要负责将作业的 DAG 转化为在分布式环境中可以执行的 Task
在 1.12 之前的 Flink 版本中，Flink 支持以下两种调度模式：
| 模式 | 特点 | 场景 |
|---|---|---|
| EAGER | 申请一个作业所需要的全部资源，然后同时调度这个作业的全部 Task，所有的 Task 之间采取 Pipeline 的方式进行通信 | Stream 作业场景 |
| LAZY | 先调度上游，等待上游产生数据或结束后再调度下游，类似 Spark 的 Stage 执行模式。 | Batch 作业场景 |
EAGER 模式
- 12 个 task 会一起调度，集群需要有足够的资源
LAZY 模式
- 最小调度一个 task 即可，集群有 1 个 slot 资源可以运行
在 1.12 后的 Flink 版本：
- 由 Pipeline 的数据交换方式连接的 Task 构成为一个 Pipeline Region；本质上，不管是流作业还是批作业，都是按照 Pipeline Region 粒度来申请资源和调度任务。
- ALL_EDGES_BLOCKING:
- 所有 Task 之间的数据交换都是 BLOCKING 模式；中间需要 shuffle 落盘才能进行下一任务
- 分为 12 个 pipeline region;
- ALL_EDGES_PIPELINED:
- 所有 Task 之间的数据交换都是 PIPELINE 模式；中间不作 shuffle 落盘
- 分为 1 个 pipeline region;
# Shuffle Service 层
Shuffle：在分布式计算中，用来连接上下游数据交互的过程叫做 Shuffle。实际上，分布式计算中所有涉及到上下游衔接的过程，都可以理解为 Shuffle。
针对不同的分布式计算框架，Shuffle 通常有几种不同的实现:
- 基于文件的 Pull Based Shuffle，比如 Spark 或 MR，它的特点是具有较高的容错性，适合较大规模的批处理作业，由于是基于文件的，它的容错性和稳定性会更好一些；
- 基于 Pipeline 的 PushBased Shuffle，比如 Flink、Storm、Presto 等，它的特点是低延迟和高高性能，但是因为 shuffle 数据没有存储下来，如果是 batch 任务的话，就需要进行重跑恢复；
流和批 Shuffle 之间的差异:
- Shuffle 数据的生命周期：流作业的 Shuffle 数据与 Task 是绑定的，而批作业的 Shuffle 数据与 Task 是解耦的；
- Shuffle 数据存储介质：流作业的生命周期比较短、而且流作业为了实时性，Shuffle 通常存储在内存中，批作业因为数据量比较大以及容错的需求，一般会存储在磁盘里；
- Shuffle 的部署方式：流作业 Shuffle 服务和计算节点部署在一起，可以减少网络开销，从而减少 latency，而批作业则不同。
- Flink 对于流和批提供内种类型的 Shuffle，虽然 Streaming 和 Batch Shuffle 在具体的策略上存任一定的差异，但本质上都是为了对数据进行 Re-Partition，因此不同的 Shuffle 之间是存在一定的共性的。 所以 Flink 的目标是提供一套统一的 Shume 架构， 即可以满足不同 Shuffle 在策略上的定制，同时还能避免在共性需求上进行重复开发。
为了统一 Flink 在 Streaming 和 Batch 模式下的 Shuffle 架构，Flink 实现了一个 Pluggable 的 Shuffle Service 框架，抽象出一些公共模块。
- 在 Streaming 和 OLAP 场景
- 为了性能的需要，通常会使用基于 Pipeline 的 Shuffle 模式
- 在 Batch 场景
- 一般会选取 目 Blocking 的 Shuffle 模式
对于 ShuffleService，Flink 开源社区已经支持：
- Netty Shuffle Service：既支持 pipeline 又支持 blocking，Flink 默认的 shuffleService 策略；
- Remote Shuffle Service：既支持 pipeline 又支持 blocking，不过对于 pipeline 模式，走 remote 反而会性能下降，主要是有用在 batch 的 blocking 场景，字节内部是基于 CSS 来实现的 RSS。
# Flink 架构优化
# 流 / 批 / OLAP 业务场景概述
在实际生产环境中对数据处理要求不同，OLAP：处理时间秒级 1-10s，用于数据分析 BA 报表等
| 模块 | 流式计算 | 批式计算 | 交互式分析 OLAP |
|---|---|---|---|
| SQL | Yes | Yes | Yes |
| 实时性 | 高、毫秒级 | 低 | 高、查询延迟秒级，但要求高并发查询 |
| 容错能力 | 高 | 中、大作业失败重跑代价高 | No，失败重试即可 |
| 状态 | Yes | No | No |
| 准确性 | Exactly Once，要求高，重跑需要回复之前的状态 | Exactly Once，失败重跑即可 | Exactly Once，失败重跑即可 |
| 扩展性 | Yes | Yes | Yes |
# OLAP
OLAP（On-Line Analytical Processing）：联机分析处理，OLAP 是数据仓库系统的主要应用，支持复杂的分析操作，侧重决策支持，并且提供直观易懂的查询结果。
OLTP（on-line transaction processing）：联机事务处理，传统的关系型数据库的主要应用，主要是基本的、日常的事务处理。
| OLTP | OLAP | |
|---|---|---|
| 用户 | 操作人员，低层管理人员 | 决策人员，高级管理人员 |
| 功能 | 日常操作处理 | 分析决策 |
| DB 设计 | 面向应用 | 面向主题 |
| 数据 | 当前的，最新的细节的，二维的分立的 | 历史的，聚集的，多维的，集成的，统一的 |
| 存取 | 读 / 写数十条记录 | 读上百万条记录 |
| 工作单位 | 简单的事务 | 复杂的查询 |
| DB 大小 | 100MB-GB | 100GB-TB |
| 时间要求 | 具有实时性 | 对时间的要求不严格 |
| 主要应用 | 数据库 | 数据仓库 |
常见 OLAP 引擎对比
# 为什么三种场景可以用一套引擎来解决
- 批式计算是流式计算的特例，Everything is Streams，有界数据集（批式数据）也是一种数据流、一种特殊的数据流；
- 而 OLAP 计算是一种特殊的批式计算，它对并发和实时性要求更高，其他情况与普通批式作业没有特别大区别。
因此，理论上，我们是可以用一套引擎架构来解决上述三种场景，只不过需要对不同场景支持相应的扩展性、并允许做不同的优化策略。
Apache Flink 从流式计算出发，需要想支持 Batch 和 OLAP 场景，就需要解决下面的问题:
Batch 场景需求：
- 流批一体支持
- Unify DataStream APl;
- Scheduler;
- Shuffle Service;
- Failover Recovery;
OLAP 场景需求：
- 短查询作业场景
- 高并发支持；
- 极致处理性能；
# Flink 如何支持 OLAP 场景
Flink 做 OLAP 的优势：
- 统一引擎：流处理、批处理、OLAP 统一使用 Flink 引擎；
- 降低学习成本，仅需要学习一个引擎；
- 提高开发效率，很多 SQL 是流批通用；
- 提高维护效率，可以更集中维护好一个引擎；
- 既有优势：利用 Flink 已有的很多特性，使 OLAP 使用场景更为广泛；
- 使用流处理的内存计算、Pipeline；
- 支持代码动态生成；
- Session 模式的 MMP 架构
- Pipeline Shuffle 也可以支持批处理数据落盘能力；
- 相互增强：OLAP 能享有现有引擎的优势，同时也能增强引擎能力
- 无统计信息场景的优化；
- 开发更高效的算子；
- 使 Flink 同时兼备流、批、OLAP 处理的能力，成为更通用的框架。
- 生态支持：
- 跨数据源查询支持
- TCP-DS 基准测试性能强
- 统一引擎：流处理、批处理、OLAP 统一使用 Flink 引擎；
Flink OLAP 场景的挑战
- 秒级和毫秒级的小作业
- 作业频繁启停，资源碎片
- Latency + QPS 的要求
# Apache Flink 的 OLAP 场景面临的问题
Flink OLAP 架构现状
- Client: 提交 SQL Query ;
- Gateway：接收 Client 提交的 SQLQuery，对 SQL 进行语法解析和查询优化，生成 Flink 作业执行计划，提交给 Session 集群；
- Session Cluster：执行作业调度及计算，并返回结果。
架构与功能模块：
- JobManager 与 ResourceManager 在一个进程内启动，无法对 JobManager 进行水平扩展；
- Gateway 与 Flink Session Cluster 互相独立，无法进行统一管理。
作业管理及部署模块：
- JobManager 处理和调度作业时，负责的功能比较多，导致单作业处理时间长，并占用了过多的内存；
- TaskManager 部署计算任务时，任务初始化部分耗时严重，消耗大量 CPU。
资源管理及计算任务调度:
- 资源申请及资源释放流程链路过长
- Slot 作为资源管理单元，JM 管理 slot 资源，导致 JM 无法感知到 TM 维度的资源分布，使得资源管理完全依赖于 ResourceManager
其他：
- 作业心跳与 Failover 机制，并不合适 AP 这种秒级或亳秒级计算场景；
- AP 目前使用 Batch 算子进行计算，这些算子初始化比较耗时；