flink架构师5-CEP、一致性、YARN

一 、 CEP 0：18~1：10

二、一致性保证 1：10 ~2：25

1. 我们使用FlinkKafkaConumser,并且启用Checkpoint，偏移量会通过checkpoint保存到state里
   面，并且默认会写入到kafka的特殊主体中，也就是__consumer_offset

2. setCommitOffsetsOnCheckpoints 默认会true，就是把偏移量写入特殊主题中

3. Flink自动重启的过程中，读取的偏移量是state中的偏移量，如果state里面没有那么从
   __consumer_offset里读取偏移量，如果__consumer_offset里面没有那么就会从earliest或者lastest读取数据

redis通过幂等性实现仅一次语义

4.1.5 写kafka保证Exactly once

两阶段提交
Flink的两阶段提交
核心源码
通过幂等性实现仅一次语义
在分布式系统中，可以使用两阶段提交来实现事务性从而保证数据的一致性，两阶段提交分为：预提交阶段与
提交阶段，通常包含两个角色：协调者与执行者，协调者用于用于管理所有执行者的操作，执行者用于执行具
体的提交操作，具体的操作流程：

1. 首先协调者会送预提交(pre-commit)命令有的执行者
2. 执行者执行预提交操作然后发送一条反馈(ack)消息给协调者
3. 待协调者收到所有执行者的成功反馈，则发送一条提交信息(commit)给执行者
4. 执行者执行提交操作
   如果在流程2中部分预提交失败，那么协调者就会收到一条失败的反馈，则会发送一条rollback消息给所有执
   行者，执行回滚操作，保证数据一致性；但是如果在流程4中，出现部分提交成功部分提交失败，那么就会造
   成数据的不一致，因此后面也提出了3PC或者通过其他补偿机制来保证数据最终一致性
   flink中两阶段提交是为了保证端到端的Exactly Once，主要依托checkpoint机制来实现，先看一下
   checkpoint的整体流程，
   1.JobManager会周期性的发送执行checkpoint命令(start checkpoint)；
   2.当source端收到执行指令后会产生一条barrier消息插入到input消息队列中，当处理到barrier时
   会执行本地checkpoint, 并且会将barrier发送到下一个节点，当checkpoint完成之后会发送一条ack信
   息给JobManager;
5. 当所有节点都完成checkpoint之后，JobManager会收到来自所有节点的ack信息，那么就表示一次
   完整的checkpoint的完成；
6. JobManager会给所有节点发送一条callback信息，表示通知checkpoint完成消息。接下来就可以
   提交事务了
   对比flink整个checkpoint机制调用流程可以发现与2PC非常相似，JobManager相当于协调者，flink提
   供了CheckpointedFunction与CheckpointListener这样两个接口，CheckpointedFunction中有
   snapshotState方法，每次checkpoint触发执行方法，通常会将缓存数据放入状态中，可以理解为是一个
   hook，这个方法里面可以实现预提交，CheckpointListener中有notifyCheckpointComplete方法，
   checkpoint完成之后的通知方法，这里可以做一些额外的操作，比如真正提交kafka的事务；在2PC中提到
   如果对应流程2预提交失败，那么本次checkpoint就被取消不会执行，不会影响数据一致性.如果流程4失
   败，那么重启从上一次的checkpoints重新计算。

三、Flink On Yarn 2：25 ~ 2：35

第一种【yarn-session.sh(开辟资源)+flink run(提交任务)】
启动一个一直运行的flink集群
/bin/yarn-session.sh -n 2 -jm 1024 -tm 1024 [-d]
•执行任务
•./bin/flink run WordCount.jar --hostname xxx --port 8888
停止任务 【web界面或者命令行执行cancel命令】

第二种【flink run -m yarn-cluster(开辟资源+提交任务)】
• 启动集群，执行任务
•./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024
./examples/batch/WordCount.jar
注意：client端必须要设置YARN_CONF_DIR或者HADOOP_CONF_DIR或者HADOOP_HOME环境变量，通过这
个环境变量来读取YARN和HDFS的配置信息，否则启动会失败

image.png

四、SQL 流平台 2：35 ~2：53

基于袋鼠云

五、作业 2：53 ~3：13