Apache Flink 刘 驰

Flink概述 Flink起源于一个叫做Stratosphere的研究项目，目标是建立下一代大数据分析引擎，其在2014年4月16日成为Apache的孵化项目，从Stratosphere 0.6开始，正式更名为Flink。 应用 请求 应用 流数据 数据库 设备 流 …… 批数据 文件/对象 存储 Apache Flink是一个面向分布式的流数据处理和批数据处理的开源计算平台，并能够基于同一个Flink Runtime，提供支持流处理和批处理两种类型应用的功能。

统一的批处理与流处理系统 Flink能够同时处理批处理任务与流处理任务，其灵活的执行引擎支持完全原生的批量的数据处理和流式的数据处理。

Flink特性 重要特性 支持Libraries 高性能 流和批的统一处理 容错机制 在JVM内部进行内存的自我管理；当处理hash、index、caching和sorting时，Flink的内存管理方式很高效 支持数据驱动的窗口、可以基于时间（event time或processing time）、count和session来构建窗口、窗口也可以定制化等 灵活的窗口处理 内存管理 重要特性 支持Libraries 高性能 机器学习、图计算、Table API等库 Flink不必做许多手动的配置、管道式(pipeline)的数据处理方式 流和批的统一处理 容错机制 通过轻量级、分布式快照 实现 Flink中批和流有各自的API，既可以开发批处理程序，也可以开发流处理程序

Flink关键技术 Deployment层：Flink支持多种部署模式：本地、集群(Standalone/YARN Cluster)和云端(GCE/EC2)。 Runtime层：提供了支持Flink计算的全部核心实现，比如：支持分布式Stream处理、JobGraph到ExecutionGraph的映射、调度等等，为上层API层提供基础服务。 API层：主要实现了面向Stream的流处理API和面向Batch的批处理API，其中面向流处理对应DataStream API，面向批处理对应DataSet API。 Libraries层：该层也可以称为Flink应用框架层，根据API层的划分，分为面向流处理和面向批处理两类： 面向流处理支持：CEP(复杂事件处理)、基于SQL-like的操作(基于Table的关系操作)等； 面向批处理支持：FlinkML(机器学习库)、Gelly(图处理)等。

Flink的分层模型 Deployment层：Flink支持多种部署模式：本地、集群(Standalone/YARN Cluster)和云端(GCE/EC2)。 Runtime层：提供了支持Flink计算的全部核心实现，比如：支持分布式Stream处理、JobGraph到ExecutionGraph的映射、调度等，为上层API层提供基础服务。 API层：主要实现了面向Stream的流处理API和面向Batch的批处理API，其中面向流处理对应DataStream API，面向批处理对应DataSet API。 Libraries层：该层也可以称为Flink应用框架层，根据API层的划分，分为面向流处理和面向批处理两类： 面向流处理支持：CEP(复杂事件处理)、基于SQL-like的操作(基于Table的关系操作)等； 面向批处理支持：FlinkML(机器学习库)、Gelly(图处理)等。

Process Function (事件，状态，时间) Flink分层API 高级分析API SQL/Table API (动态表) val stats = stream .keyBy("sensor") .timeWindow(Time.seconds(5)) .sum((a, b) -> a.add(b)) 流和批处理 DataStream API (流、窗口) 基于状态事件驱动的应用 Process Function (事件，状态，时间) def processElement(event: MyEvent, ctx: Context, out: Collector[Result]) = { // work with event and state (event, state.value) match { … } out.collect(…) // emit events state.update(…) // modify state // schedule a timer callback ctx.timerService.registerEventTimeTimer(event.timestamp + 500) }

从JobGraph到ExcecutionGraph JobGraph是一个Job的用户逻辑视图表示，将一个用户要对数据流进行的处理表示为单个DAG图。JobVertex表示对数据流进行的转换操作，比如map、flatMap、filter等，IntermediateDataSet是由上游的JobVertex生成，作为下游JobVertex的输入。 ExecutionGraph是JobGraph的并行表示，也就是实际JobManager调度一个Job在TaskManager上运行的逻辑视图，它也是一个DAG图。

Flink执行流程 Flink系统主要包括三个组件：JobClient、JobManager和TaskManager。 3. JobManager负责资源的协调和Job的执行； 4. 一旦资源分配完成，task就会分配到不同的TaskManager； 5. TaskManager会初始化线程去执行task，并根据程序的执行状态向JobManager反馈，执行的状态包括starting、in progress、finished以及canceled和failing等； 6. 当Job执行完成，结果会返回给JobClient。

Flink执行流程 左上子图：有2个TaskManager，每个TaskManager有3个Task Slot 左下子图：一个Flink Job，逻辑上包含了1个data source、1个MapFunction、1个ReduceFunction，对应一个JobGraph 左下子图：用户提交的Flink Job对各个Operator进行的配置—data source的并行度设置为4，MapFunction的并行度也为4，ReduceFunction的并行度为3，在JobManager端对应于ExecutionGraph 右上子图：TaskManager 1上，有2个并行的ExecutionVertex组成的DAG图，它们各占用一个Task Slot 右下子图：TaskManager 2上，也有2个并行的ExecutionVertex组成的DAG图，它们也各占用一个Task Slot 在2个TaskManager上运行的4个Execution是并行执行的

JobMananger Master进程，负责Job的管理和资源的协调。包括任务调度、检查点管理、失败恢复等。 JobManager包含三个重要的组件： Actor系统 调度器 检查点 对于集群HA模式，可以同时多个master进程，其中一个作为leader，其他作为standby。当leader失败时，会选出一个standby的master作为新的leader（通过zookeeper实现leader选举）。 

Actor系统 Flink内部使用Akka模型作为JobManager和TaskManager之间的通信机制 Actor系统是个容器，包含许多不同的Actor，分别扮演着不同的角色。Actor系统提供类似于调度、配置、日志等服务，同时包含了所有actors初始化时的线程池。 Akka是一个构建在JVM上，基于Actor模型的的并发框架，为构建伸缩性强，有弹性的响应式并发应用提高更好的平台。

所有的Actors存在着层级的关系：新加入的Actor会被分配一个父类的Actor。每个父类的Actor都负责监控其子类Actor，当子类Actor出现错误时，自己先尝试重启并修复错误；如果子类Actor无法修复，则将问题交给父类Actor处理。  Actors之间的通信采用一个消息系统：每个Actor都有一个“邮箱”，用于读取消息。如果Actors是本地的，则消息在共享内存中共享；如果Actors是远程的，则消息通过RPC远程调用 Actor是一个有状态和行为的容器，其线程持续处理从“邮箱”中接收到的消息。也就是说，Actor中的状态和行为则由收到的消息决定。

调度器 Flink中的Executors被定义为task slots（线程槽位）。每个Task Manager需要管理一个或多个task slots。 Flink通过SlotSharingGroup和CoLocationGroup来决定哪些task需要被共享，哪些task需要被单独的slot使用。

检查点 Flink的检查点机制是保证其一致性容错功能的骨架，其持续地为分布式的数据流和有状态的operator生成一致性的快照。 该机制改良自Chandy-Lamport算法，叫做ABS（轻量级异步Barrier快照），具体参见论文：  Lightweight Asynchronous Snapshots for Distributed Dataflows

Barrier作为一种Event，是Flink快照中最主要的元素。 Barrier会随着data record一起被注入到流数据中，而且不会超越data record。每个barrier都有一个唯一的ID，将data record分到不同的检查点的范围中。上图展示了barrier被注入到data record中的过程。 每个快照中的状态都会报告给Job Manager的检查点协调器； 快照发生时，flink会在某些有状态的operator上对data record进行对齐操作（alignment），目的是避免失败恢复时重复消费数据。这个过程也是exactly once的保证。 Flink的容错机制持续的构建轻量级的分布式快照，因此负载非常低。通常，这些有状态的快照都被放在HDFS中存储（state backend）。程序一旦失败，Flink将停止executor并从最近的完成了的检查点开始恢复（依赖可重发的数据源+快照）。 对齐操作的时间仅是毫秒级的。但是对于某些极端的应用，在每个operator上产生的毫秒级延迟也不能允许的话，则可以选择降级到at least once，即跳过对齐操作，当失败恢复时可能发生重复消费数据的情况。Flink默认采用exactly once处理。

TaskMananger Task Managers是具体执行tasks的worker节点，执行发生在一个JVM中的一个或多个线程中。 Task的并行度是由运行在Task Manager中的task slots的数量决定。如果一个Task Manager有4个slots，那么JVM的内存将分配给每个task slot 25%的内存。一个Task slot中可以运行1个或多个线程，同一个slot中的线程又可以共享相同的JVM。

Job Client Job Client并不是Flink程序执行中的内部组件，而是程序执行的入口。 Job Client负责接收用户提交的程序，并创建一个data flow，然后将生成的data flow提交给Job Manager。一旦执行完成，Job Client将返回给用户结果。 Data flow就是执行计划，比如一个简单的word count的程序：

DataFlow 从source到map的data flow，是一对一的关系，不产生shuffle操作；而从map到groupBy操作，flink会根据key将数据重分布，即shuffle操作，目的是聚合数据，产生正确的结果。

并行的流DataFlow 默认情况下，Flink的data flow都是分布式并行处理的。 对于数据的并行处理，flink将operators和数据流进行partition。Operator partitions叫做sub-tasks。数据流又可以分为一对一的传输与重分布的情况。

Flink应用场景 Athena X Streaming SQL Platform Service Streaming Platform as a Service Fraud detection Streaming Analytics Platform Bouyues, 建筑与房地产公司 100s jobs, 1000s nodes, TBs state metrics, analytics, real time ML Streaming SQL as a platform Huawei Cloud provides Cloud Service based on Flink running 30 production applications powered by Flink and processing 10 billion raw events per day

商业案例 --- 美团外卖-Flink实践 资源隔离 故障容错 生态建设 1.Yarn打标签，节点物理隔离 隔离策略 容灾 封装Kafka读写组件 1.Yarn打标签，节点物理隔离 2.离线DataNode与实时计算节点的隔离 1.JobManager HA 2.多机房 3.流热备 1.屏蔽地址 2.权限管理 3.指标监控

商业案例 --- 美团外卖-Flink平台化 作业管理 监控报警 调优诊断 1.实时计算引擎提供统一的日志和指标方案； 2.为业务提供按条件过滤的日志检索； 3.为业务提供自定义时间跨度的指标查询； 4.基于日志和指标，为业务提供可配置报警。

商业案例 --- 美团外卖-Flink应用 Petra实时指标聚合 MLX机器学习平台 业务场景特征 大规模机器学习平台 1. 模型训练 1. 多业务维度聚合 2.低延迟、秒级结果输出 3.对晚到数据的容忍能力 4.基于业务时间(Event Time) 1. 模型训练 2. 特征提取 3. 在线预估

Flink其他案例 阿里巴巴 King游戏公司 Bouygues电信公司 Zalando电子商务公司 1.将复杂的转换操作转化为相对简单的; 2.确保分析终端用户可以更快的访问数据。 King游戏公司 Bouygues电信公司 Zalando电子商务公司 使用Flink监控其有线和无线网络： 1.用于网络/传感器检测和错误检测； 2.实现快速故障响应。 所有基础设施团队使用Flink实时更新产品细节和库存信息： 1.优化电子商务的实时搜索结果； 2.为用户提供更高的关联性。 使用Flink-powered数据分析平台提供实时数据分析，从游戏数据中大幅缩短观察时间。