Flink05--状态,联结,容错

端到端的有效一次必须经过开始和结束点+有效一次

将要输出的数据写出到日志中当checkpoint触发的时候，再将日志中的数据写出到外部存储（只取决于checkPoint）瑕疵:成功一半，失败一半不能提供百分百端到端的Exactly-Once

barrier对齐

可以进行恢复数据.

sink

Chandy-Lamport(分布式快照)

At-least-once(最少一次)

//运行环境StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();environment.setParallelism(2);//启用检查点environment.enableCheckpointing(5000);//设置  状态后端-->嵌入的RocksDB状态后端(嵌入的RocksDB状态后端)environment.setStateBackend(new EmbeddedRocksDBStateBackend());//远程状态备份  getCheckpointConfig(得到配置文件)->setCheckpointStorage设置检查点存储environment.getCheckpointConfig().setCheckpointStorage(\"hdfs://node02:8020/flink/checkpoints\");//数据源DataStreamSource<String> source = environment.socketTextStream(\"localhost\

有状态算子这么管理呢？

事务操作要么全部完成提交 commit，要么全部失败回滚 rollback如果你的输出端具备了这一特性当数据写入到数据库的时候并不会马上提交数据，直到checkpoint被触发并且barrier传递到sink端的时候才会被提交，这样就把barrier和事务绑定到了一起瑕疵:有可能检查点完成了，但是事务提交失败有可能事务提交成功，但是检查点后续操作失败事务写的方式能提供端到端的Exactly-Once一致性，它的代价也是非常明显的，就是牺牲了延迟。输出数据不再是实时写入到外部系统，而是分批次地提交。目前来说，没有完美的故障恢复和Exactly-Once保障机制，对于开发者来说，需要在不同需求之间权衡。

自动托管

定义

Transformation<Operator>

确保状态的更新和访问存储介质:1.存储在JVM或者Fs中2.存储在RocksDB中

WAL预写日志

class YjxxtStateBackendFunction implements font color=\"#ff0000\

利用窗口机制按照指定字段和（滚动/滑动/会话）窗口进行inner join

容错机制

确保相同的数据在一个窗口中

font color=\"#ff0000\

At-least-once + 幂等

一个算子任务会按照并行度分为多个并行子任务执行不同的slot在计算资源上是物理隔离的

No-keyed

端到端的Sink前提: 要防止数据重复写入1.文本型的存储 txt word2.数据库型的存储mysql oracle

font color=\"#0000ff\

算子的State + CheckPoint一次 checkpoint 后所持久化的各算子的状态数据，确保是经过了相同数据的影响

Local State Management本地状态管理

间隔的“上界” (upperBound)和“下界“(lowerBound)

无界数据或事件的连续处理流或事件处理应用程序可以被描述为有向图每个边表示数据或事件流，每个顶点表示运算符，会使用程序中定义的逻辑处理来自相邻边的数据或事件。

记录所有进程的状态和所有管道的状态-->直接拍摄整个照片(类似)

Flink 2.0

运行流程

精确一次&有效一次没有引擎能够保证正好只处理一次，处理一次+恢复一次=2次同一个数据即使处理多次得到的效果还是相同的

2PS两阶段提交

托管读取

数据处理语义

可能会有数据丢输入端只需要将数据进行发送，不管后续接受端是否接受到

//执行环境        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();        environment.setParallelism(2);        //设置启动检查点        environment.enableCheckpointing(5000);        //配置检查点文件，设置检查点存贮位置        environment.getCheckpointConfig().setCheckpointStorage(\"file:///\" + System.getProperty(\"user.dir\") + File.separator + \"ckpt\");        //数据源        DataStreamSource<String> source = environment.socketTextStream(\"localhost\

合并流

手动托管

Flink 1.0

Offset偏移量

有状态算子保存在什么位置呢？

名词

有状态算子有哪些呢？

Interval join

在受到barrier信息后保存到状态后端的偏移量.

由Flink统一管理的，状态的存储访问、故障恢复和重组等一系列问题都由Flink实现.当应用发生横向扩展时，状态也会自动地重组分配到所有的子任务实例上。Flink提供了值状态(ValueState) 、列表状态(ListState) 、射状态(MapState) 、聚合状态 (AggregateState) 等多种结构，内部支持各种数据类型。程序员只需要调用接口即可

SavePoint保存点

//配置程序执行的参数        Configuration configuration = new Configuration();        configuration.setString(\"execution.savepoint.path\

算法演化

//开始进行流的JoininfoStream.coGroup(priceStream)        .where(i -> i.f0)        .equalTo(p -> p.f0)        .window(TumblingEventTimeWindows.of(Time.seconds(10)))        .font color=\"#ff0000\

意义

Connet

Exactly-once(只被处理一次)

算子状态

有状态

无状态

就是开辟了一块内存，我们自己管理，实现状态的序列化和故障恢复。Flink不会对状态进行任何自动操作，也不知道状态的具体数据类型，只会把它当作最原始的字节(Byte) 数组来存储程序员需要花费大量的精力来处理状态的管理和维护。

有状态算子保存内存怎么处理？

状态类型

自定义检查点函数

托管保存

特定时间点记录下来的分布式系统的全局状态.(检查点)

Keyed

有可能重复处理数输入端将数据进行发送，接受端没有接受(ACK) 到则再次进行发送

检查点流程

Remote State Checkpointing远程状态备份

At-most-once(最多一次)

状态后端

HDFS路径，其格式为“hdfs://nameservice/flink/checkpoints”

Barrier数据屏障

Checkpoint 的主要目的是为意外失败的作业提供恢复机制。生命周期由 Flink 管理Savepoint 由用户创建，拥有和删除。 他们的用例是计划的，手动备份和恢复。

inner join内连接分类Tumbling-Sliding-Session只有相同窗口的数据才会进行计算

CheckPoint(检查点)

窗口联结

//开始进行流的JoininfoStream.keyBy(info -> info.f0)        .intervalJoin(priceStream.keyBy(price -> price.f0))        font color=\"#ff0000\

相同的数据会被保留.

Outter Join外连接 除了输出匹配的元素对以外，未能匹配的元素也会输出。第一个迭代器为join左边流

模型简化

流程

端到端的STS

@Overridepublic void initializeState(FunctionInitializationContext context) throws Exception {    StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.seconds(60))            .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)            .setTtlTimeCharacteristic(StateTtlConfig.TtlTimeCharacteristic.ProcessingTime)            .cleanupFullSnapshot()            .build();    //创建对象的描述器    ListStateDescriptor<Integer> descriptor = new ListStateDescriptor<Integer>(\"CountListState\

端到端的Source前提: 能够进行重复消费KafkaSource--能够记录偏移量、能够重放数据、将偏移量记录在 state 中

Chandy-Lamport

TTL

管理方式

匹配

join

多个stream

分类

按照计算资源隔离此时可以按照每一个算子的font color=\"#ff0000\

托管恢复

Source

产生

分布式快照

At-least-once + 去重

CoGroup

Union

幂等操作在编程中一个幕等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。绝对值: Math.abs(num) -- 执行一次和执行100次效果相同利用数据结构的特点，保证数据的幂等性

目的

等到上游所有的并行子分区barrier都到齐，才去保存当前任务的状态缺点：先到达的分区要做缓存等待，会造成数据堆积（背压）

Unaligned Checkpoint未对齐的检查点

将本地的 state 的备份到远程的存储介质上1.内存2.分布式文件系统

保存到HDFS

机制革新(反压问题)

End-to-End Exactly-Once(端到端的有效一次)

数据反压问题

两阶段提交(Two-phase Commit，简称2PC)Flink消费到Kafka数据之后，就会开启一个Kafka的事务 (协调者) ，正常写入Kafka分区日志标记但未提交，也就是预提交 (Per-commit)一旦所有的Operator完成各自的Per-commit，他们会发起一个commit操作如果有任意一个Per-commit失败，所有其他的Per-commit必须停止，并且Flink会回滚到最近成功完成的CheckPoint当所有的Operator完成任务时，Sink端就收到checkpoint barrier(检查点分界线)，Sink保存当前状态，存入Checkpoint，通知JobManager，并提交外部事物，用于提交外部检查点的数据JobManager收到所有任务通知，发出确认信息，表示Checkpoint已经完成，Sink收到JobManager的确认信息，正式提交这段时间的教据外部系统 (Kafka) 关闭事务，提交的数据可以正常消费了