企业官网建设流程全解析

一、检查点（Checkpoint）

1、定义

将之前某个时间点所有的状态保存下来，这份存档就是所谓的检查点。遇到故障重启时，我们可以从最近一次成功保存的检查点中读档，恢复出之前的状态。

2、检查点的保存

1、周期性的触发保存，间隔时间可以进行设置
2、应该在所有任务都恰好处理完一个相同的输入数据的时候保存它们的状态。这样故障时只需要让source任务向数据源重新提交保存的偏移量，请求数据重放就可以了。

3、检查点算法

采用了基于Chandy-Lamport算法的分布式快照，可以在不暂停整体流处理的前提下，将状态备份保存到检查点。

1、检查点分界线（Barrier）
借鉴水位线设计，在数据流中插入一个特殊的数据结构，专门用来表示触发检查点保存的时间点。收到保存检查点的指令后，source任务可以在当前数据流中插入这个结构，之后的所有任务只要遇到它就开始对状态做持久化快照保存。由于数据流是保持顺序依次处理的，因此遇到这个标示的任务就代表之前的数据都处理完了，可以保存到检查点。他之后的数据就需要保存到下一个检查点。这个特殊的数据结构，把一条流上的数据按照不同的检查点分隔开，所以叫检查点的分界线。
在JobManager中有一个“检查点协调器”，专门用来协调处理检查点的相关工作。检查点协调器会定期向TaskManager发出指令，要求保存检查点（带着检查点ID），TaskManager会让所有的source任务把自己的偏移量（算子状态）保存，每一个任务做完状态持久化存储之后，会向JobManager确认检查点完成，并将带有检查点ID的分界线插入到当前的数据流中，像正常的数据一样像下游传递，之后source任务就可以继续读入新的数据处理了。

2、barrier对齐的精确一次
watermark指示的是“之前的数据全部到齐了”，而barrier指示的是“之前所有数据的状态更改保存到当前检查点”。当上游任务向多个并行下游任务发送barrier时需要广播出去，而当多个上游任务向同一个下游任务传递分界线时，需要在下游任务执行分界线对齐操作，也就是需要等到所有上游并行分区的相同ID barrier都到齐，才可以开始状态的保存。

具体过程：
1、触发检查点：JobManager向source发送Barrier；
2、向下游广播发送Barrier
3、Barrier对齐：下游需要收到上游所有并行度传递过来的Barrier才做自身状态的保存；
4、状态保存：有状态的算子将状态保存持久化
5、先处理等待对齐时的缓存数据（barrier先到达的分区等待对齐时需要将barrier后面到来的数据缓存处理，也就是后面的数据不计算阻塞等待，一定程度上会影响处理速度），然后继续正常处理新到的数据

3、barrier对齐的至少一次
与精确一次唯一的区别就是在：barrier先到达的分区等待对齐时，barrier后面到来的数据不放入缓存，而是直接进行计算并且更新状态，等到barrier对齐后做状态的备份。这样在故障恢复时，在barrier对齐间到达的数据就会再重复计算一遍，所以只能保证至少一次

4、非barrier对齐的精确一次
flink 1.11出的新算法，与barrier对齐的精确一次的区别就是在：当第一个barrier到达输入缓冲区时
（1）直接将barrier放到输出缓冲区末端，向下游传递
（2）标记数据：一是被第一个barrier越过的输入缓冲区和输出缓冲区的数据；二是在其他barrier之前的所有数据
（3）把标记数据和状态一起保存到检查点中，恢复时这些数据也一起恢复到对应的位置
总结：
一个task收到第一个barrier时，就开始执行上面的备份步骤，其后面的数据接着计算输出；未到的其他barrier，前面的数据备份，同时接着计算输出；最后一个barrier到达该task时，这个task的检查点保存结束。

4、检查点配置

barrier对齐

//创建执行环境StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();//启用检查点：默认是barrier对齐的，周期为5s，精确一次env.enableCheckpointing(5000,CheckpointingMode.EXACTLY_ONCE);CheckpointConfigcheckpointConfig=env.getCheckpointConfig();//指定检查点的存储位置checkpointConfig.setCheckpointStorage("hdfs://node-1:8020/chk");//Checkpoint的超时时间：默认10分钟checkpointConfig.setCheckpointTimeout(60000);//同时运行中的checkpoint的最大数量checkpointConfig.setMaxConcurrentCheckpoints(2);//最小等待间隔：上一轮checkpoint结束到下一轮checkpoint开始之间的间隔，设置了>0，并发就会变成1checkpointConfig.setMinPauseBetweenCheckpoints(1000);//取消作业时，checkpoint的数据是否保留在外部系统;删除：DELETE_ON_CANCELLATION(主动cancel时，删除存在外部系统的检查点文件，如果程序突然挂掉不会删）checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//保留//允许checkpoint连续失败的次数，默认0，表示checkpoint一失败，job就挂掉checkpointConfig.setTolerableCheckpointFailureNumber(2);

barrier非对齐

//创建执行环境StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();//启用检查点：默认是barrier对齐的，周期为5s，精确一次env.enableCheckpointing(5000,CheckpointingMode.EXACTLY_ONCE);CheckpointConfigcheckpointConfig=env.getCheckpointConfig();//同时运行中的checkpoint的最大数量checkpointConfig.setMaxConcurrentCheckpoints(1);//开启非对齐检查点，要求：Checkpoint模式必须是精确一次，最大并发数必须设为1checkpointConfig.enableUnalignedCheckpoints();/*设置启用非对齐超时时间 * 1.要开启非对齐检查点才生效：默认0，表示一开始就直接用非对齐的检查点 * 2.如果大于0，一开始则用对齐的检查点（barrier对齐），对齐的时间超过这个参数，自动切换成非对齐检查点（barrier非对齐） */checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));

二、保存点（Savepoint）

1、保存点和检查点的区别

1、保存点的原理和算法与检查点完全相同，只是多了一些额外的元数据。
2、保存点和检查点之间最大的区别就是触发时机。检查点由flink自动管理，定期创建，发生故障后自动读取进行恢复；而保存点不会自动创建，必须由用户明确地手动触发保存操作。

2、保存点的用途

可以当作一个强大的运维工具来使用。我们可以在需要的时候创建一个保存点，然后停止应用，做出一些调整之后再从保存点重启。
具体场景：
1、版本管理和归档存储
2、更新flink版本
3、更新应用程序
4、调整并行度
5、暂停应用程序

注意：保存点能够在程序更改的时候依然兼容，前提是状态的拓扑结构和数据类型不变。保存点中状态都是以算子ID-状态名称这样的key-value组织起来的，算子id可以在代码中不同算子后直接调用.uid()来指定。对于没有设置id的算子，flink默认会自动进行设置，所以在重新启动应用后可能会导致id不同而无法兼容以前的状态，所以为了方便后续的维护，强烈建议在程序中为每一个算子手动指定id

3、使用保存点

1、创建保存点

#在命令行中执行bin/flink savepoint :jobId[保存点的存储路径]#创建保存点之后同时停掉作业bin/flink stop--savepointPath[保存点的存储路径]:jobId

保存点的默认存储路径可以在flink-conf.yaml中配置

state.savepoints.dir:文件路径

2、从保存点重启应用

bin/flink run-s :savepointPath[runArgs]#yarn的运行模式bin/flink run-s :savepointPath-yid application-id[runArgs]

3、使用保存点切换状态后端
使用savepoint恢复状态时，也可以更新状态后端。注意不要在代码中指定状态后端了，而是通过配置文件或提交时的参数来配置。

具体步骤：

#提交flink作业./bin/flink run-application-d-t yarn-application-Dstate.backend=hashmap-c 全类名 jar包路径#停止flink作业时触发保存点,stop优雅停止，但是要求source实现StoppableFunction接口bin/flink stop--savepointPath[保存点的存储路径]job-id-yid application-id#或者cancel停止bin/flink cancel--savepointPath[保存点的存储路径]job-id-yid application-id#从savepoint恢复作业，同时修改状态后端./bin/flink run-application-d-t yarn-application-s 保存点的存储路径-Dstate.backend=rocksdb-c 全类名 jar包路径#从checkpoint恢复作业，不支持修改状态后端./bin/flink run-application-d-t yarn-application-s 检查点的存储路径（到编号文件名）-c 全类名 jar包路径

三、端到端精确一次

实际应用中最希望，也最难做到的一致性语义就是端到端精准一次。flink内部检查点机制可以保证故障后数据不丢（数据能够重放的前提下），并且只处理一次。关键在于输入的数据源端和输出的外部存储器端是否能做到精准一次。

1、输入端

数据可重放，如Kafka可重置读取数据偏移量

2、flink处理

开启checkpoint且精准一次
1、barrier对齐精准一次
2、非barrier对齐精准一次

3、输出端

1、幂等写入

就是说一个操作可以重复执行很多次，但只导致一次结果更改，后面再重复执行就不会对结果起作用了。
比如：1、利用MySQL的主键upsert
2、hbase的rowkey唯一

2、事务写入

事务的四个基本特性：原子性，一致性，隔离性和持久性。当sink任务遇到barrier时，开始保存状态的同时就开启一个事务，接下来所有数据的写入都在这个事务中；待到当前检查点保存完毕时，将事务提交，所有写入的数据就真正可用了。如果中间过程出现故障，状态会回退到上一个检查点，而当前事务没有正常关闭，所以也会回滚，写入到外部的数据就会被撤销了
1、预写日志（WAL）（不常用）
让不支持事务的存储系统来实现事务写入
①先把结果数据作为日志（log）状态保存起来
②进行检查点保存时，也会将这些结果数据一并做持久化存储
③在收到检查点完成的通知时，将所有结果一次性写入外部系统
④在成功写入所有数据后，在内部再次确认数据是否成功写入外部系统，将确认信息也进行持久化保存，这才代表着检查点的真正完成

注意：如果数据成功写入外部系统，但是确认信息丢失，就会认为这次检查点是失败的，如果发生故障，就会回滚到前一个检查点，就会导致数据重复，所以它只能保证数据至少一次

2、两阶段提交
用来解决分布式场景下的事务，先预提交，等检查点完成之后再正式提交，他需要外部系统提供事务支持
① 当第一条数据来，或者收到检查点的分界线时，sink任务都会启动一个事务
②接下来收到的所有数据，都通过这个事务写入外部系统，但是不可用，是预提交状态
③当sink任务收到job manager发来的检查点完成的通知时，正式提交事务，写入的结果就真正可用了。

注意：如果中间发生故障时，当前未提交的事务就会回滚，写入外部系统的数据就撤回。数据写入是流式的，不再有预写日志时批处理的性能问题。对外部系统有很高的要求。如下：
①外部系统必须提供事务支持，或sink任务必须能够模拟外部系统上的事务
②在检查点间隔期间，必须能够开启一个事务并接受数据写入
③在收到检查点完成的通知之前，事务必须是等待提交的状态，在故障恢复的情况下，这可能需要一些时间，如果这个时候外部系统关闭事务（如超时了），那么未提交的数据就会丢失
④sink任务必须能够在进程失败后恢复事务
⑤提交事务必须是幂等操作，也就是说，事务的重复提交应该是无效的

4、flink sink到kafka的精确一次实例

//创建执行环境StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();//代码中用到hdfs，需要导入hadoop依赖，指定访问hdfs的用户名System.setProperty("HADOOP_USER_NAME","haha");//启用检查点：默认是barrier对齐的，周期为5s，精确一次env.enableCheckpointing(5000,CheckpointingMode.EXACTLY_ONCE);CheckpointConfigcheckpointConfig=env.getCheckpointConfig();//指定检查点的存储位置checkpointConfig.setCheckpointStorage("hdfs://node-1:8020/chk");checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//保留//读取kafkaKafkaSource<String>kafkaSource=KafkaSource.builder().setBootstrapServers("node-1:9092,node-2:9092,node-3:9092")//指定kafka节点的地址和端口.setGroupId("test")//指定消费者组的id.setTopics("hadoop")//指定消费的Topic.setValueOnlyDeserializer(newSimpleStringSchema())//指定反序列化器，这个是只反序列化value.setStartingOffsets(OffsetsInitializer.latest())//flink消费kafka的策略.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")//设置事务的隔离级别，读已提交；默认是读未提交，也就是预提交的数据也会读到.build();DataStreamSource<String>kafkadata=env.fromSource(kafkaSource,WatermarkStrategy.noWatermarks,"kafkasource");//fromSource(Souurce的实现类，Watermark，名字）//写出到kafkaKafkaSink<String>kafkaSink=KafkaSink.<String>builder().setBootstrapServers("node-1:9092,node-2:9092,node-3:9092")//指定kafka节点的地址和端口.setRecordSerializer(KafkaRecordSerializationSchema.<String>builder().setTopic("hadoop1").setValueSerializationSchema(newSimpleStringSchema()).build())//指定序列化器：指定topic名称，具体的序列化.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)//写到kafka的一致性级别：精确一次,开启两阶段提交.setTransactionalIdPrefix("flink-")//如果是精确一次，必须设置事务的前缀.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,10*60*1000+"")//如果是精准一次，必须设置事务超时时间：大于checkpoint间隔，小于max 15分钟.build();kafkadata.sinkTo(kafkaSink);env.execute();