2019 大数据实时计算  流计算
Blink 阿里内部 Flink 分支版本

核心概念和基础  整体架构 核心概念 算子等
进阶          数据传输 容错机制 序列化 数据热点 反压等 
源码          核心代码实现 Job提交流程 数据交换 分布式快照机制 FlinkSQL 



基础篇


Flink简介

分布式处理引擎 用于对无界和有界数据流进行有状态计算
提供 数据分布 容错机制及资源管理等功能

DataSet    API 对静态数据进行批处理，将静态数据抽象成分布式数据集
DataStream API 对数据流进行流处理 ，将流式数据抽象成分布式数据流
Table API      对结构化数据进行查询操作，将结构化数据抽象成关系表，通过类SQL的DSL对关系表进行各种查询操作

特定领域的库
FlinkML   Flink 机器学习库  
Gelly     Flink图计算库
CEP       复杂事件处理

支持高吞吐 低延迟 高性能的流处理
支持带有事件时间的窗口 （Window） 操作
支持高度灵活的窗口 （Window） 操作，支持基于 time  count session 以及 data-driven 的窗口操作
支持有状态计算的 Exactly-once 语义
支持具有 Backpressure 功能的持续流模型
支持基于轻量级分布式快照（Snapshot）实现的容错
运行时同时支持 Batch onStreaming 处理和 Streaming 处理
Flink 在 JVM 内部实现了自己的内存管理
支持迭代计算
支持程序自动优化 避免特定情况下 Shuffle 排序等昂贵操作，可缓存中间结果 



Flink 与 Spark 比较

Flink 是实时处理引擎，基于事件驱动
Spark Streaming  微批（Micro-Batch） 


架构模型

Spark Streaming 运行时主要角色
Master Worker Driver Executor
Flink 运行时主要包含  Jobmanager Taskmanager和Slot



任务调度

Spark Streaming 连续不断的生成微小的数据批次，构建有向无环图DAG
依次创建 DStreamGraph JobGenerator JobScheduler 

Flink 根据用户提交的代码生成 StreamGraph，经过优化生成 JobGraph
然后提交给 JobManager进行处理，JobManager 根据 JobGraph 生成 ExecutionGraph
ExecutionGraph 是 Flink 调度最核心的数据结构
JobManager 根据 ExecutionGraph 对 Job 进行调度


时间机制

Spark Streaming 只支持处理时间
Flink 支持 处理时间 事件时间 注入时间 支持 watermark机制来处理滞后数据


容错机制

Spark Streaming 任务，可设置 checkpoint
可以从上次 checkpoint 之处恢复，可能会重复处理，不能做到恰好一次处理语义

Flink使用两阶段提交协议来解决该问题




Flink集群角色及作用


Flink 程序 运行时主要有 TaskManager JobManager Client三种角色
JobManager 集群 Master ，整个集群的协调者，
负责接收 Job，协调检查点，Failover 故障恢复等 ，同时管理集群从节点TaskManager

TaskManager 负责执行计算的Worker ，执行 Job的一组Task
每个TaskManager负责管理其所在节点上的资源信息，如内存 磁盘 网络
启动时候将资源的状态向JobManager汇报



Client
提交一个Flink程序时，首先创建一个Client
对提交的Flink程序进行预处理 ，然后提交给JobManager



Task Slot
 
TaskManager 将节点上管理的资源分为不同的Slot,固定大小的资源子集
避免不同Job的Task互相竞争内存资源，Slot只做内存隔离,没有做CPU隔离



Flink 常用算子

Map Filter KeyBy Reduce Window



分区策略

分区策略用来决定数据如何发送至下游

GlobalPartitioner     
ShufflePartitioner    
RebalancePartitioner  
RescalePartitioner    
BroadcastPartitioner   
ForwardPartitioner   
KeyGroupStreamPartitioner    
CustomPartitionerWrapper  用户自定义分区器 实现Partitioner接口 

public interface Partitioner< K> extends java.io.Serializable, Function {
    int partition(K key, int numPartitions);
}



并行度设置

任务并行执行

可以从四个不同层面设置并行度

操作算子层面(Operator Level)
执行环境层面(Execution Environment Level)
客户端层面(Client Level)
系统层面(System Level)

优先级 
算子层面 > 环境层面 > 客户端层面 > 系统层面



Slot和parallelism

slot是指taskmanager的并发执行能力
taskmanager.numberOfTaskSlots 配置为3  
那么每个 taskmanager 中分配3个 TaskSlot
3个 taskmanager 一共有9个TaskSlot

parallelism.default 

parallelism 并行度 并发度




重启策略

固定延迟重启策略（Fixed Delay Restart Strategy）
故障率重启策略（Failure Rate Restart Strategy）
没有重启策略（No Restart Strategy）
Fallback重启策略（Fallback Restart Strategy）


Flink中的分布式缓存
taskmanager 本地缓存，避免重复读取数据

org.apache.flink.api.java.ExecutionEnvironment 
public void registerCachedFile(String filePath, String name) 
public void registerCachedFile(String filePath, String name, boolean executable)

val env = ExecutionEnvironment.getExecutionEnvironment
// register a file from HDFS
env.registerCachedFile( "hdfs:///path/to/your/file", "hdfsFile")
// register a local  file 
env.registerCachedFile( "file:///path/to/exec/file", "localExecFile", true)


Flink中的广播变量
可理解为一个共享变量，可以把一个dataset 数据集广播出去
然后不同的task在节点上都能够获取到，在每个节点上只会存在一份


窗口

time-tumbling-window 
不重叠 timeWindow(Time.seconds(5))

time-sliding-window 
有重叠 timeWindow(Time.seconds(5), Time.seconds(3))


count-tumbling-window
countWindow(5)

count-sliding-window 
countWindow(5,3)


org.apache.flink.streaming.api.datastream.KeyedStream.timeWindow(Time, Time)
public WindowedStream< T, KEY, TimeWindow> timeWindow(Time size, Time slide) 

DataStream< WordWithCount> windowCounts = ...
            .keyBy("word")
            .timeWindow(Time.seconds(5), Time.seconds(1))
            .reduce(new ReduceFunction< WordWithCount>() {
                public WordWithCount reduce(WordWithCount a, WordWithCount b) {
                    return new WordWithCount(a.word, a.count + b.count);
                }
            });
            
            

状态存储

计算过程存储中间状态 避免数据丢失和状态恢复
影响状态持久化  checkpoint  保存点
MemoryStateBackend FsStateBackend RocksDBStateBackend


流处理中的时间

事件时间 摄入时间 处理时间
EventTime       EventTimeWindow
IngesingtTime   IngestingTimeWindow    以 source 的systemTime为准
ProcessingTime  ProcessingTimeWindow 以 operator 的systemTime 为准


Watermark 概念和作用

Watermark 是 为了处理 EventTime 窗口计算提出的一种机制, 本质上是一种时间戳
Watermark 和 Window 一起用来处理乱序事件


Flink Table & SQL   TableEnvironment的作用

TableEnvironment作用

在内部catalog中注册表
注册外部catalog
执行SQL查询
注册用户定义（标量，表或聚合）函数
将DataStream或DataSet转换为表
持有对ExecutionEnvironment或StreamExecutionEnvironment的引用


Flink SQL

SQL解析基于 Apache Calcite

SQL解析过程

使用 SQL 开发业务应用
用calcite对SQL进行语法检验
转换成逻辑树节点 ,最终形成逻辑计划
采用Flink自定义的优化规则和calcite火山模型、启发式模型共同对逻辑树进行优化，生成最优的物理计划
对物理计划采用janino codegen生成代码，生成用低阶API描述的流应用
提交执行



进阶篇



如何支持批流一体
批处理是流处理的一种特殊情况
DataSet API 和 DataStream API
DataSet API 会被废弃
DataStream API  可设置成批处理模式




如何做到高效的数据交换

数据在不同的task中进行交换
TaskManager 负责
先从缓冲buffer中收集records，然后再发送
批次发送



如何容错

State   CheckPoint  savepoint
State 存储计算过程中的中间状态
Checkpoint  定时快照 备份状态


分布式快照原理
根据Chandy-Lamport算法
持续创建分布式数据流及其状态的一致快照
核心思想是在 input source 端插入 barrier，
控制 barrier 的同步来实现 snapshot 的备份和 exactly-once 语义



如何保证Exactly-once语义

幂等 和 事务 两阶段提交
两阶段提交和状态保存来实现端到端的一致性语义
开始事务（beginTransaction）创建一个临时文件夹，写入数据
预提交（preCommit）将内存中缓存的数据写入文件并关闭
正式提交（commit）将之前写完的临时文件放入目标目录下 
丢弃（abort）丢弃临时文件
若失败发生在预提交成功后，正式提交前。可以根据状态来提交预提交的数据，也可删除预提交的数据。



kafka 连接器
独立的connector模块
 

 
 
内存管理机制
MemorySegment
堆外内存利用
数据超出内存限制，溢写到磁盘
避免序列化开销 直接操作字节

Network Buffers 
TaskManager启动时分配，每个块是32K，默认分配2048个
taskmanager.network.numberOfBuffers

Memory Manage pool
MemorySegment块，用于运行时的算子（Sort/Join/Shuffle等）

User Code
除了Memory Manager之外的内存用于User code和TaskManager本身的数据结构






序列化

Java自带的序列化和反序列化 占内存 效率低 性能不高

flink序列化机制
包含 类型描述符，泛型类型提取和类型序列化框架
TypeInformation  
支持以下几种类型

BasicTypeInfo       任意Java 基本类型或 String 类型
BasicArrayTypeInfo  任意Java基本类型数组或 String 数组
WritableTypeInfo    任意 Hadoop Writable 接口的实现类
TupleTypeInfo       任意的 Flink Tuple 类型  支持Tuple1 to Tuple25 
CaseClassTypeInfo   任意的 Scala CaseClass(包括 Scala tuples)
PojoTypeInfo        任意的 POJO (Java or Scala) 
GenericTypeInfo     任意无法匹配之前几种类型的类
针对前六种类型数据集，Flink可自动生成对应的TypeSerializer，高效地对数据集进行序列化和反序列化




Window数据倾斜如何处理
数据进入窗口前做预聚合
重新设计窗口聚合的key



使用聚合函数 GroupBy Distinct KeyBy 等出现数据热点如何处理

在业务上规避这类问题  单独处理热点数据
Key设计  热点key拆分
参数设置  
Flink 1.9.0 SQL(Blink Planner) 性能优化中一项重要的改进就是升级了微批模型，即 MiniBatch
缓存一定的数据后再触发处理，以减少对State的访问，从而提升吞吐和减少数据的输出量



Flink任务延迟高如何解决

后台任务管理
查看哪个算子和task出现了反压
最主要的手段是资源调优和算子调优

资源调优
对作业中的Operator的并发数（parallelism） CPU（core） 堆内存（heap_memory）等参数进行调优

作业参数调优
并行度设置 State设置 checkpoint设置




反压机制

内部基于 producer-consumer 模型进行消息传递，
反压基于该模型
高效有界的分布式阻塞队列
下游消费慢，阻塞上游 



Flink和Strom反压机制区别

Storm 通过监控 Bolt 中的接收队列负载情况
超过高水位值就会将反压信息写到 Zookeeper 
Zookeeper 上的 watch 会通知该拓扑的所有 Worker 都进入反压状态，最后 Spout 停止发送 tuple

Flink反压使用高效有界的分布式阻塞队列，下游消费慢，阻塞上游 

Flink逐级反压，Storm直接从源头降速



Operator Chains（算子链）
尽可能地将operator的subtask链接（chain）在一起形成task
每个task在一个线程中执行
减少线程之间的切换，减少消息的序列化/反序列化，减少数据在缓冲区的交换，减少延迟的同时提高整体的吞吐量

算子链 算子融合



Operator chain 算子链 形成条件

上下游的并行度一致
下游节点的入度为1  
上下游节点都在同一个 slot group 中 
下游节点的 chain 策略为 ALWAYS（可以与上下游链接 map flatmap filter等默认是ALWAYS）
上游节点的 chain 策略为 ALWAYS 或 HEAD（只能与下游链接，不能与上游链接，Source默认是HEAD）
两个节点间数据分区方式是 forward 
没有禁用 chain



Flink1.9新特性

支持hive读写 支持UDF
Flink SQL TopN和GroupBy等优化
Checkpoint跟savepoint针对实际业务场景做了优化
Flink state查询



消费kafka数据时如何处理脏数据

处理前加一个fliter算子, 过滤掉脏数据


源码篇


Job提交流程

Job 转化成DAG任务运行

StreamGraph JobGraph ExecutionGraph
JobManager与Client的交互基于Akka，消息驱动
整个Flink Job的提交还包含 ActorSystem的创建，JobManager的启动，TaskManager的启动和注册



三层图结构
StreamGraph JobGraph ExecutionGraph

StreamGraph 
最接近代码所表达的逻辑层面的计算拓扑结构
按照用户代码的执行顺序向StreamExecutionEnvironment添加StreamTransformation构成流式图

JobGraph 
从StreamGraph生成，将可以串联合并的节点进行合并，
设置节点之间的边，安排资源共享slot槽位和放置相关联的节点，
上传任务所需的文件，设置检查点配置等
相当于经过部分初始化和优化处理的任务图。

ExecutionGraph 
由JobGraph转换而来，包含了任务具体执行所需的内容，是最贴近底层实现的执行图



JobManger作用

负责整个 Flink 集群任务的调度以及资源的管理
从客户端中获取提交的应用，然后根据集群中 TaskManager 上 TaskSlot 的使用情况
为提交的应用分配相应的 TaskSlot 资源并命令 TaskManager 启动 应用

相当于整个集群的 Master 节点
JobManager 和 TaskManager 之间通过 Actor System 通信
获取任务执行的情况并通过 Actor System 将应用的任务执行情况发送给客户端

任务执行过程中， JobManager 会触发 Checkpoint 操作
每个 TaskManager 节点 收到 Checkpoint 指令后，完成 Checkpoint 操作
所有的 Checkpoint 协调过程都在 Fink JobManager 中完成

任务完成后，Flink 将任务执行的信息反馈给客户端，并且释放掉 TaskManager 中的资源 




JobManger在集群启动过程中的作用

接收Flink作业，调度Task，收集作业状态和管理TaskManager
包含一个Actor，完成如下操作
RegisterTaskManage
SubmitJob
CancelJob
UpdateTaskExecutionState
由TaskManager发送，用来更新执行节点(ExecutionVertex)的状态 

RequestNextInputSplit
TaskManager上的Task请求下一个输入split，成功则返回NextInputSplit，否则返回null

JobStatusChanged
作业状态变化(RUNNING, CANCELING, FINISHED等) 



TaskManager作用

相当于整个集群的 Slave 节点，负责具体的任务执行和对应任务在每个节点上的资源申请和管理

客户端提交任务给 JobManager
JobManager 根据已注册的 TaskManager 资源情况
将任务分配给TaskManager节点，然后启动并运行任务

TaskManager 使用 Slot 资源启动 Task，  TaskManager 之间的数据交互 
多个任务和 Task 之间通过 TaskSlot 方式共享系统资源
TaskManager 管理多个 TaskSlot 资源池


启动流程
org.apache.flink.runtime.taskmanager.TaskManager
selectNetworkInterfaceAndRunTaskManager 
启动后向JobManager注册，注册完成后，进行部分模块的初始化




计算资源调度机制
Task slot
多个task运行在同一个JVM中



数据抽象及数据交换过程

MemorySegment
32kb 内存块 
堆内或堆外内存

Buffer StreamRecord



分布式快照机制

分布式数据流和操作算子状态的一致性快照
一致性checkpoint，在发生故障时回滚
基于Chandy-Lamport算法


barriers在数据流源处被注入并行数据流中
快照n的barriers被插入的位置（称之为Sn）是快照所包含的数据在数据源中最大位置
例如，在Kafka中，此位置是分区中最后一条记录的偏移量
将该位置报告给checkpoint协调器（JobManager）
然后barriers向下游流动  当一个中间操作算子从其所有输入流中收到快照n的barriers时
它会为快照n发出barriers进入其所有输出流中
一旦sink操作算子（流式DAG的末端）从其所有输入流接收到barriers n，它就向checkpoint协调器确认快照n完成
在所有sink确认快照后，意味快照着已完成

一旦完成快照n job将永远不再向数据源请求Sn之前的记录




FlinkSQL实现


SQL 校验 解析以及 优化交给 Apache Calcite
SQL query 经过 Calcite 解析器转变成 SQL 节点树
通过验证后构建成 Calcite 的抽象语法树

Table API 上的调用构建成 Table API 的抽象语法树
通过 Calcite 提供的 RelBuilder 转变成 Calcite 的抽象语法树
然后依次转换成逻辑执行计划和物理执行计划



 小巧的Java编译器 Janino  
 flink流处理WordCount 实例