第8章
分流
1.使用侧输出流
2.合流
2.1 union :使用 ProcessFunction 处理合流后的数据
2.2 Connect :
两条流的格式可以不一样, map操作使用CoMapFunction,process 传入:CoProcessFunction
2.2 BroadcastConnectedStream
keyBy 进行了按键分区,那么要传入的就是 KeyedBroadcastProcessFunction;
如果没有按键分区,就传入 BroadcastProcessFunction
3.基于时间的合流——双流联结(Join)
3.1 窗口联结(Window Join)
stream1.join(stream2)
.where(<KeySelector>)
.equalTo(<KeySelector>)
.window(<WindowAssigner>)
.apply(<JoinFunction>)
3.2 间隔联结(Interval Join)
所以匹配的条件为:a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound
process函数传入:ProcessJoinFunction
3.3 窗口同组联结(Window CoGroup)
stream1.coGroup(stream2)
.where(<KeySelector>)
.equalTo(<KeySelector>)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.apply(<CoGroupFunction>)
第九章:状态编程
1 状态的分类:托管状态(Managed State)和原始状态(Raw State)
1.托管状态分为两类:算子状态(Operator State)和按键分区状态(Keyed State)
1.1算子状态可以用在所有算子上,使用的时候其实就跟一个本地变量没什么区别——因为本地变量的作用域也是当前任务实例。在使用时,我们还需进一步实现 CheckpointedFunction 接口。
ListState、UnionListState 和 BroadcastState
1.2 按键分区状态(Keyed State):状态是根据输入流中定义的键(key)来维护和访问的,所以只能定义在按键分区流(KeyedStream)中,也就 keyBy 之后才可以使用
支持的数据结构:值状态(ValueState)、列表状态(ListState)、映射状态(MapState)、归约状态(ReducingState)、聚合状态(AggregatingState)
open中声明状态:getRuntimeContext.getMapState(new MapStateDescriptor[String,String]("my-map-state",classOf[String],classOf[String]))
2.状态生存时间(TTL)
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(10))//这就是设定的状态生存时间
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)//创建状态和更改状态(写操作)时更新失效时间
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)//表示从不返回过期值
.build();
ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("my
state", String.class);
stateDescriptor.enableTimeToLive(ttlConfig);
3.状态持久化和状态后端
1. 状态后端的分类:“哈希表状态后端”(HashMapStateBackend)、内嵌 RocksDB 状态后端”(EmbeddedRocksDBStateBackend)。
第十章:检查点(Checkpoint)文章来源:https://www.toymoban.com/news/detail-657240.html
1.从检查点来恢复状态了。具体的步骤为:
(1)重启应用,所有任务的状态会清空
(2)读取检查点,重置状态。找到最近一次保存的检查点,从中读出每个算子任务状态的快照,分别填充到对应的状态中。
(3)重放数据:保存检查点后开始重新读取数据,这可以通过 Source 任务向外部数据源重新提交偏移量(offset)来实现
(4)继续处理数据
2.检查点算法:
Flink 使用了 Chandy-Lamport 算法的一种变体,被称为“异步分界线快照”(asynchronous barrier snapshotting)算法。算法的核心就是两个原则:当上游任务向多个并行
下游任务发送 barrier 时,需要广播出去;而当多个上游任务向同一个下游任务传递 barrier 时,
需要在下游任务执行“分界线对齐”(barrier alignment)操作,也就是需要等到所有并行分区
的 barrier 都到齐,才可以开始状态的保存。
具体过程如下:
(1)JobManager 发送指令,触发检查点的保存;Source 任务保存状态,插入分界线
(2)状态快照保存完成,分界线向下游传递
(3)向下游多个并行子任务广播分界线,执行分界线对齐
(4)分界线对齐后,保存状态到持久化存储
(5)先处理缓存数据,然后正常继续处理
3 端到端精确一次(end-to-end exactly-once)
3.1 输入端保证
外部数据源就必须拥有重放数据的能力
3.2输出端保证
幂等写入
事务写入:预写日志(WAL)和两阶段提交(2PC)
(1)预写日志(write-ahead-log,WAL):缺点:再次确认可能会导致数据写出成功,但是确认消息失败,导致的数据重复写入
①先把结果数据作为日志(log)状态保存起来
②进行检查点保存时,也会将这些结果数据一并做持久化存储
③在收到检查点完成的通知时,将所有结果一次性写入外部系统。
(2)两阶段提交(two-phase-commit,2PC)文章来源地址https://www.toymoban.com/news/detail-657240.html
具体的实现步骤为:
①当第一条数据到来时,或者收到检查点的分界线时,Sink 任务都会启动一个事务。
②接下来接收到的所有数据,都通过这个事务写入外部系统;这时由于事务没有提交,所
以数据尽管写入了外部系统,但是不可用,是“预提交”的状态。
③当 Sink 任务收到 JobManager 发来检查点完成的通知时,正式提交事务,写入的结果就
真正可用了。
当中间发生故障时,当前未提交的事务就会回滚,于是所有写入外部系统的数据也就实现
了撤回
2PC 对外部系统的要求
外部系统必须提供事务支持,或者 Sink 任务必须能够模拟外部系统上的事务。
⚫ 在检查点的间隔期间里,必须能够开启一个事务并接受数据写入。
⚫ 在收到检查点完成的通知之前,事务必须是“等待提交”的状态。在故障恢复的情况
下,这可能需要一些时间。如果这个时候外部系统关闭事务(例如超时了),那么未
提交的数据就会丢失。
⚫ Sink 任务必须能够在进程失败后恢复事务。
⚫ 提交事务必须是幂等操作。也就是说,事务的重复提交应该是无效的。
(3) kafka-flink-kafka 实现端到端 exactly-once 的具体过程可以分解如下
1.启动检查点保存:标志着我们进入了两阶段提交协议的“预提交”阶段
2.算子任务对状态做快照保存到状态后端
3.Sink 任务开启事务,进行预提交
4.检查点保存完成,提交事务
当所有算子的快照都完成,JobManager 会向所有任务发确认通知,告诉大家当前检查点已成功保存,当 Sink 任务收到确认通知后,就会正式提交之前的事务
需要的配置:必须启用检查点、 FlinkKafkaProducer 的构造函数中传入参数 Semantic.EXACTLY_ONCE、Kafka 读取数据的消费者的隔离级别(read_committed)、事务超时配置
到了这里,关于Flink分流,合流,状态,checkpoint和精准一次笔记的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!