山东网站建设公司推荐,wordpress 存储位置,个人网页设计模版,西安seo站内优化Flink学习笔记 前言#xff1a;今天是学习 flink 的第 12 天啦#xff01;学习了 flink 高级特性和新特性之 End-to-End Exactly-Once#xff08;端到端精确一致性语义#xff09;#xff0c;主要是解决大数据领域数据从数据源到数据落点的一致性#xff0c;不会容易造成…Flink学习笔记 前言今天是学习 flink 的第 12 天啦学习了 flink 高级特性和新特性之 End-to-End Exactly-Once端到端精确一致性语义主要是解决大数据领域数据从数据源到数据落点的一致性不会容易造成数据丢失的问题结合自己实验猜想和代码实践总结了很多自己的理解和想法希望和大家多多交流 Tips端到端的一致性语义说明每一步都算术每一天的努力都不会白费明天也要继续努力 文章目录 Flink学习笔记四、Flink 高级特性和新特性1. End-to-End Exactly-Once1.1 流处理的数据语义1.1.1 At most once最多一次1.1.2 At least once至少一次1.1.3 Exactly once精确一次1.1.4 End to End Exactly once端到端精确一次1.1.5 流计算系统如何支持一次性语义(1) At least once 去重(2) At least once 幂等(3) 分布式快照(4) 方法汇总 1.2 End-to-End Exactly-Once 实现1.2.1 Source1.2.2 Transformation1.2.3 sink 1.3 Flink Kafka 的 End-to-End Exactly Once1.3.1 版本声明1.3.2 两阶段提交-API1.3.3 两阶段提交-流程 1.4 案例演示1.4.1 Flink Kafka 实现 End-to-End Exactly Once1.4.2 Flink MySQL 实现 End-to-End Exactly Once 四、Flink 高级特性和新特性 1. End-to-End Exactly-Once 1.1 流处理的数据语义 顺序At most once最多一次 At least once至少一次 Exactly once精确一次 End to End Exactly once端到端一次 1.1.1 At most once最多一次 最简单的恢复方式直接从失败的下个数据恢复程序丢失刚刚失败的数据。 1.1.2 At least once至少一次 由于事件是可以重传的可能造成数据重复。 1.1.3 Exactly once精确一次 依赖 checkpoint 机制回滚恢复数据保持所有记录仅影响内部状态一次即不考虑部分数据泄露到下游。 1.1.4 End to End Exactly once端到端精确一次 Flink 应用从 Source 端开始到 Sink 端结束保持所有记录影响内部和外部状态一次即考虑部分数据泄露到下游。 1.1.5 流计算系统如何支持一次性语义 (1) At least once 去重 (2) At least once 幂等 (3) 分布式快照 (4) 方法汇总 Exactly Once 实现方式优点缺点At least once 去重1. 故障对性能的影响是局部的2. 故障的影响不一定随着拓扑大小而增加1. 可能需要大量的存储和基础设施来支持2. 每个算子的每个事件都有资源开销At least once 幂等1. 实现简单开销较低1. 依赖存储特性和数据特征分布式快照1. 较小的性能和资源开销1. barrier 同步2. 任何算子发生故障都需要全局暂停和状态回滚3. 拓扑越大对性能的潜在影响越大 1.2 End-to-End Exactly-Once 实现 1.2.1 Source 发生故障时需要支持重设数据的读取位置如Kafka可以通过offset来实现其他的没有offset系统可以自己实现累加器计数 1.2.2 Transformation 分布式快照机制 同 Spark 相比Spark 仅仅是针对 Driver 的故障恢复 Checkpoint而 Flink 的快照可以到算子级别并且对全局数据也可以做快照Flink 的分布式快照受到 Chandy-Lamport 分布式快照算法启发同时进行了量身定做。 Barrier 数据栅栏是一个标记不会干扰正常数据处理一个数据源可以有多个 barrier多个数据源快流等慢流。 异步和增量 异步快照不会阻塞任务增量快照每次进行的全量快照是根据上一次更新的。 1.2.3 sink 幂等写入 任意多次向一个系统写入数据只对目标系统产生一次结果影响。key和 value 可以控制不重复 事务写入 借鉴数据库的事务机制结合自身 checkpoint 机制 分阶段快照先保存数据不向外部系统提交checkpoint 确认过上下游一致后才向外部系统 commit。 实现方式 1- 预写日志Write-Ahead-LogWAL 通用性强但不能保证百分比因为要写入内存这个易失介质。 2- 两阶段提交Two-Phase-Commit2PC 如果外部系统自身支持事务比如MySQL、Kafka可以使用2PC方式百分百端到端的Exactly-Once。 缺点 牺牲了延迟输出不是实时写入而是分批写入 1.3 Flink Kafka 的 End-to-End Exactly Once 1.3.1 版本声明 Flink 1.4 版本之前支持 Exactly Once 语义仅限于应用内部。 Flink 1.4 版本之后通过两阶段提交 (TwoPhaseCommitSinkFunction) 支持 End-To-End Exactly Once而且要求 Kafka 0.11。 1.3.2 两阶段提交-API 实现方法封装在抽象类TwoPhaseCommitSinkFunction 重写方法 beginTransaction 开启事务前在目标文件系统的临时目录中创建一个临时文件处理数据时将数据写入此文件 preCommit 在预提交阶段刷写flush文件然后关闭文件之后就不能写入到文件了将为下一检查点的任何后续写入启动新事务 commit 在提交阶段将预提交的文件原子性移动到真正的目标目录中注意会增加输出数据可见性的延迟 abort 在中止阶段删除临时文件。 1.3.3 两阶段提交-流程 1- 数据源阶段 对接数据源系统 2- 预提交阶段pre-commit-内部状态 Flink 开始 checkpoint就会进入 pre-commit 阶段同时 JobManager 的 Coordinator 会将 Barrier 注入数据流中 3- 预提交阶段pre-commit-外部状态 当所有的 barrier 在算子中成功进行一遍传递(就是 Checkpoint 完成)并完成快照后则“预提交”阶段完成 4- commit 阶段 所有算子完成“预提交”就会发起一个commit“提交”动作任何一个“预提交”失败都会回滚到最近的 checkpoint 1.4 案例演示 1.4.1 Flink Kafka 实现 End-to-End Exactly Once 例子1普通方式——内部一致性语义重点在生产者 API 设置上只是简单序列化为字节流 SimpleStringSchema package cn.itcast.day12.endtoend;import org.apache.commons.lang3.SystemUtils; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import org.apache.flink.streaming.connectors.kafka.internals.KafkaSerializationSchemaWrapper; import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper; import org.apache.flink.util.Collector; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig;import java.util.Properties;/*** author lql* time 2024-03-07 14:51:04* description TODOtopictest3 终端生产生数据控制台打印 topictest4数据*/ public class Kafka_Flink_Kafka_EndToEnd_ExactlyOnce {public static void main(String[] args) throws Exception {//todo 1初始化flink流处理环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(5000);//todo 2判断当前的环境env.setStateBackend(new HashMapStateBackend());if(SystemUtils.IS_OS_WINDOWS || SystemUtils.IS_OS_MAC){env.getCheckpointConfig().setCheckpointStorage(file:///D:\\checkpoint);}else{env.getCheckpointConfig().setCheckpointStorage(args[0]);}//todo 3设置checkpoint的其他参数//设置checkpoint的超时时间env.getCheckpointConfig().setCheckpointTimeout(2000L);//同一个时间只能有一个栅栏在运行env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//设置checkpoint的执行模式。仅执行一次env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//设置checkpoint最小时间间隔env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);//todo 4接入数据源//指定topic的名称String topicName test;//实例化kafkaConsumer对象Properties props new Properties();props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, node1:9092);props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, test001);props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, latest); // 消费最新的数据props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); // 自动提交偏移量offsetprops.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 2000); // 提交偏移量的时间间隔props.setProperty(flink.partition-discovery.interval-millis, 5000);//开启一个后台线程每隔5s检测一次kafka的分区情况FlinkKafkaConsumerString kafkaSource new FlinkKafkaConsumer(topicName, new SimpleStringSchema(), props);//在开启checkpoint以后offset的递交会随着checkpoint的成功而递交从而实现一致性语义默认就是truekafkaSource.setCommitOffsetsOnCheckpoints(true);DataStreamSourceString kafkaDS env.addSource(kafkaSource);//todo 5单词计数操作SingleOutputStreamOperatorTuple2String, Integer wordAndOne kafkaDS.flatMap(new FlatMapFunctionString, Tuple2String, Integer() {Overridepublic void flatMap(String value, CollectorTuple2String, Integer out) throws Exception {String[] words value.split( );for (String word : words) {out.collect(Tuple2.of(word, 1));}}});//todo 6单词分组操作SingleOutputStreamOperatorTuple2String, Integer result_1 wordAndOne.keyBy(t - t.f0).sum(1);//todo 7打印计算结果result_1.print();SingleOutputStreamOperatorString result result_1.map(new MapFunctionTuple2String, Integer, String() {Overridepublic String map(Tuple2String, Integer value) throws Exception {return value.f0 _ value.f1;}});result.printToErr();//todo 8创建kafka的生产者实例//指定topic的名称String distTopicName test1;//实例化FlinkKafkaProducer对象Properties distProps new Properties();distProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, node1:9092);FlinkKafkaProducerString myProducer new FlinkKafkaProducerString(distTopicName,new SimpleStringSchema(),distProps);// 容错//todo 4将数据写入到kafkaresult.addSink(myProducer);//todo 8启动作业env.execute();} }结果在 node1 的 kafka 生产者模式终端输入数据到 test词频统计结果写入到 topictest1但不保证外部一致性语义 例子2超级方式——内部外部一致性语义 package cn.itcast.day12.endtoend;import org.apache.commons.lang3.SystemUtils; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper; import org.apache.flink.util.Collector; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig;import java.util.Properties;/*** author lql* time 2024-03-07 14:51:04* description TODOtopictest3 终端生产生数据控制台打印 topictest4数据*/ public class Kafka_Flink_Kafka_EndToEnd_ExactlyOnce_pro {public static void main(String[] args) throws Exception {//todo 1初始化flink流处理环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 在这里就不能开启了因为 kafka//env.enableCheckpointing(5000);//todo 2判断当前的环境env.setStateBackend(new HashMapStateBackend());if(SystemUtils.IS_OS_WINDOWS || SystemUtils.IS_OS_MAC){env.getCheckpointConfig().setCheckpointStorage(file:///D:\\checkpoint);}else{env.getCheckpointConfig().setCheckpointStorage(args[0]);}//todo 3设置checkpoint的其他参数//设置checkpoint的超时时间env.getCheckpointConfig().setCheckpointTimeout(2000L);//同一个时间只能有一个栅栏在运行env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//设置checkpoint的执行模式。仅执行一次env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//设置checkpoint最小时间间隔env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);//todo 4接入数据源//指定topic的名称String topicName test;//实例化kafkaConsumer对象Properties props new Properties();props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, node1:9092);props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, test001);props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, latest); // 消费最新的数据props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); // 自动提交偏移量offsetprops.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 2000); // 提交偏移量的时间间隔props.setProperty(flink.partition-discovery.interval-millis, 5000);//开启一个后台线程每隔5s检测一次kafka的分区情况FlinkKafkaConsumerString kafkaSource new FlinkKafkaConsumer(topicName, new SimpleStringSchema(), props);//在开启checkpoint以后offset的递交会随着checkpoint的成功而递交从而实现一致性语义默认就是truekafkaSource.setCommitOffsetsOnCheckpoints(true);DataStreamSourceString kafkaDS env.addSource(kafkaSource);//todo 5单词计数操作SingleOutputStreamOperatorTuple2String, Integer wordAndOne kafkaDS.flatMap(new FlatMapFunctionString, Tuple2String, Integer() {Overridepublic void flatMap(String value, CollectorTuple2String, Integer out) throws Exception {String[] words value.split( );for (String word : words) {out.collect(Tuple2.of(word, 1));}}});//todo 6单词分组操作SingleOutputStreamOperatorTuple2String, Integer result_1 wordAndOne.keyBy(t - t.f0).sum(1);//todo 7打印计算结果result_1.print();SingleOutputStreamOperatorString result result_1.map(new MapFunctionTuple2String, Integer, String() {Overridepublic String map(Tuple2String, Integer value) throws Exception {return value.f0 _ value.f1;}});result.printToErr();//todo 8创建kafka的生产者实例//指定topic的名称String distTopicName test1;//实例化FlinkKafkaProducer对象Properties distProps new Properties();distProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, node1:9092);FlinkKafkaProducerString myProducer new FlinkKafkaProducerString(distTopicName,new KeyedSerializationSchemaWrapperString(new SimpleStringSchema()),distProps,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);//todo 4将数据写入到kafkaresult.addSink(myProducer);//todo 8启动作业env.execute();} }结果在 node1 的 kafka 生产者模式终端输入数据到 test词频统计结果写入到 topictest1保证了内外部一致性语义 总结 在普通模式设置下需要提前开启 checkpoint 模式在超级模式设置下不要提前开启 checkpoint 模式不然写不进数据在超级模式设置下不是简单序列化而是事务写入 new KeyedSerializationSchemaWrapper(new SimpleStringSchema()),FlinkKafkaProducer.Semantic.EXACTLY_ONCE 1.4.2 Flink MySQL 实现 End-to-End Exactly Once 例子读取 socket 数据写入 MySQL 数据库删除数据库数据也能继续累加结果实现端到端一致性。 SQL建表 create table test.t_wordcount (word varchar(255) not null primary key,counts int default 0 null );代码 package cn.itcast.day12.endtoend;import org.apache.commons.lang3.SystemUtils; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.typeutils.base.VoidSerializer; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction; import org.apache.flink.util.Collector;import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.sql.SQLException;public class Kafka_Flink_MySQL_EndToEnd_ExactlyOnce {public static void main(String[] args) throws Exception {//todo 1初始化flink流处理环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//todo 2如果实现端对端一次性语义必须要开启checkpointenv.enableCheckpointing(5000L);//todo 3判断当前的环境env.setStateBackend(new HashMapStateBackend());if(SystemUtils.IS_OS_WINDOWS || SystemUtils.IS_OS_MAC){env.getCheckpointConfig().setCheckpointStorage(file:///D:\\checkpoint);}else{env.getCheckpointConfig().setCheckpointStorage(args[0]);}//todo 4设置checkpoint的其他参数//设置checkpoint的超时时间env.getCheckpointConfig().setCheckpointTimeout(2000L);//同一个时间只能有一个栅栏在运行env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//设置checkpoint的执行模式。仅执行一次env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//设置checkpoint最小时间间隔env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);//todo 5接入数据源读取文件获取数据DataStreamSourceString lines env.socketTextStream(node1, 9999);//todo 3数据处理// 3.1使用flatMap对单词进行拆分SingleOutputStreamOperatorString words lines.flatMap(new FlatMapFunctionString, String() {Overridepublic void flatMap(String line, CollectorString out) throws Exception {String[] words line.split( );//返回数据for (String word : words) {out.collect(word);}}});// 3.2对拆分后的单词进行记一次数SingleOutputStreamOperatorTuple2String, Integer wordAndOne words.map(new MapFunctionString, Tuple2String, Integer() {Overridepublic Tuple2String, Integer map(String word) throws Exception {return Tuple2.of(word, 1);}});// 3.3使用分组算子对key进行分组KeyedStreamTuple2String, Integer, String grouped wordAndOne.keyBy(t - t.f0);// 3.4对分组后的key进行聚合操作SingleOutputStreamOperatorTuple2String, Integer sumed grouped.sum(1);//todo 6将消费到的数据实时写入mysqlsumed.addSink(new MysqlTwoPhaseCommitSink());//todo 7运行作业env.execute();}/*** 通过两端递交的方式实现数据写入mysql*/public static class MysqlTwoPhaseCommitSink extends TwoPhaseCommitSinkFunctionTuple2String, Integer, ConnectionState, Void {public MysqlTwoPhaseCommitSink() {super(new KryoSerializer(ConnectionState.class, new ExecutionConfig()), VoidSerializer.INSTANCE);}/*** 每条数据执行一次该方法* param connectionState* param value* param context* throws Exception*/Overrideprotected void invoke(ConnectionState connectionState, Tuple2String, Integer value, Context context) throws Exception {System.err.println(start invoke.......);Connection connection connectionState.connection;// 插入一条记录但如果该记录的主键或唯一键已经存在则更新该记录。PreparedStatement pstm connection.prepareStatement(INSERT INTO t_wordcount (word, counts) VALUES (?, ?) ON DUPLICATE KEY UPDATE counts ?);pstm.setString(1, value.f0);pstm.setInt(2, value.f1);pstm.setInt(3, value.f1);// 插入数据一定是 executeUpdatepstm.executeUpdate();pstm.close();//手动制造异常if(value.f0.equals(hive)) {System.out.println(1/0);}}/*** 开启事务* return* throws Exception*/Overrideprotected ConnectionState beginTransaction() throws Exception {System.out.println( beginTransaction... );Class.forName(com.mysql.jdbc.Driver);//closing inbound before receiving peers close_notify链接地址中追加参数useSSLfalseConnection connection DriverManager.getConnection(jdbc:mysql://node1:3306/test?characterEncodingUTF-8useSSLfalse, root, 123456);connection.setAutoCommit(false);return new ConnectionState(connection);}/*** 预递交* param connectionState* throws Exception*/Overrideprotected void preCommit(ConnectionState connectionState) throws Exception {System.out.println(start preCommit...);}/*** 递交操作* param connectionState*/Overrideprotected void commit(ConnectionState connectionState) {System.out.println(start transaction...);Connection connection connectionState.connection;try {connection.commit();connection.close();} catch (SQLException e) {throw new RuntimeException(提交事物异常);}}/*** 回滚操作* param connectionState*/Overrideprotected void abort(ConnectionState connectionState) {System.out.println(start abort...);Connection connection connectionState.connection;try {connection.rollback();connection.close();} catch (SQLException e) {throw new RuntimeException(回滚事物异常);}}}static class ConnectionState {// transient 的变量能被忽略序列化private final transient Connection connection;ConnectionState(Connection connection) {this.connection connection;}} }结果mysql 数据库中删除数据后再次添加数据后仍会叠加数据。 总结 1- 两段递交自定义 sink 中 需要继承 TwoPhaseCommitSinkFunction2- kyro 序列化连接状态VoidSerializer 需要接上 INSTANCE 作为 Void 的序列化3- 开启事务时要放弃自动提交4- transient 的变量能被忽略序列化此处用于连接变量5- 数据库插入计算时要使用 executeUpdate