原理+实践|Exactly-once系列实践之KafkaToKafka

程序源代码

共 15461字,需浏览 31分钟

 ·

2023-03-11 12:56

全网最全大数据面试提升手册!

推荐阅读:


文章目录
一、Kafka输入输出流工具类
二、统计字符个数案例
三、消费者消费kafka的事务数据
四、总结与可能出现的问题

一、Kafka输入输出流工具类

代码如下(示例):

//获取kafkaStream流
    public static <T> DataStream<T> getKafkaDataStream(ParameterTool parameterTool,Class<? extends DeserializationSchema> clazz,StreamExecutionEnvironment env) throws IllegalAccessException, InstantiationException {
        //加入到flink的环境全局配置中,后续可以通过上下文获取该工具类,总而得到想要的值
        env.getConfig().setGlobalJobParameters(parameterTool);


        //kafka配置项
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", parameterTool.get("bootstrap.servers"));
        properties.setProperty("group.id",parameterTool.get("group.idsource"));
        properties.setProperty("auto.offset.reset",parameterTool.get("auto.offset.reset"));
        properties.setProperty("enable.auto.commit",parameterTool.get("enable.auto.commit", String.valueOf(false)));
        String topics = parameterTool.get("Consumertopics");

        //序列化类实例化
        DeserializationSchema<T> deserializationSchema = clazz.newInstance();

        FlinkKafkaConsumer<T> flinkKafkaConsumer = new FlinkKafkaConsumer<>(topics, deserializationSchema, properties);

        flinkKafkaConsumer.setStartFromEarliest();
        //开启kafka的offset与checkpoint绑定
        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(true);

        return env.addSource(flinkKafkaConsumer);
    }



    //获取kafka生产者通用方法

    /**
     * offsets.topic.replication.factor 用于配置offset记录的topic的partition的副本个数
     * transaction.state.log.replication.factor 事务主题的复制因子
     * transaction.state.log.min.isr 覆盖事务主题的min.insync.replicas配置
     *
     * num.partitions 新建Topic时默认的分区数
     *
     * default.replication.factor 自动创建topic时的默认副本的个数
     *
     *
     *
     * 注意:这些参数,设置得更高以确保高可用性!
     *
     * 其中 default.replication.factor 是真正决定,topi的副本数量的
     * @param parameterTool
     * @param kafkaSerializationSchema
     * @param <T>
     * @return
     */
    public static <T> FlinkKafkaProducer<T> getFlinkKafkaProducer(ParameterTool parameterTool,KafkaSerializationSchema<T> kafkaSerializationSchema){
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", parameterTool.get("bootstrap.servers"));
        properties.setProperty("group.id",parameterTool.get("group.idsink"));
//        properties.setProperty("transaction.max.timeout.ms",parameterTool.get("transaction.max.timeout.ms"));
        properties.setProperty("transaction.timeout.ms",parameterTool.get("transaction.timeout.ms"));
        properties.setProperty("client.id""flinkOutputTopicClient");
        String topics = parameterTool.get("Producetopice");

        return new FlinkKafkaProducer<T>(topics,kafkaSerializationSchema,properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

    }

注意点事项

一、消费者注意项

  1. flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(true),将kafka自动提交offset关闭并且与flink的CheckPoint绑定
  2. bootstrap.servers kafka的broker host
  3. setStartFromEarliest()设置kafka的消息消费从最初位置开始

二、生产者注意项

  1. transaction.timeout.ms 默认情况下Kafka Broker 将transaction.max.timeout.ms设置为15分钟,我们需要将此值设置低于15分钟
  2. FlinkKafkaProducer.Semantic.EXACTLY_ONCE设置kafka为精确一次

二、统计字符个数案例

代码如下(示例):

public static void main(String[] args) throws Exception {
        //1.创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.设置并行度
        env.setParallelism(4);

        //3.设置CK和状态后端
        CkAndStateBacked.setCheckPointAndStateBackend(env,"FS");

        //4.获取kafkaStream流
        InputStream kafkaPropertiesStream = KafkaToKafkaExacitly.class.getClassLoader().getResourceAsStream("kafka.properties");
        ParameterTool parameterTool=ParameterTool.fromPropertiesFile(kafkaPropertiesStream);
        //将配置流放到全局flink运行时环境
        env.getConfig().setGlobalJobParameters(parameterTool);
        SimpleStringSchema simpleStringSchema = new SimpleStringSchema();
        Class<? extends SimpleStringSchema> stringSchemaClass = simpleStringSchema.getClass();
        DataStream<String> kafkaDataStream = KafkaUtil.getKafkaDataStream(parameterTool, stringSchemaClass, env);

        System.out.println("==================================================");
        kafkaDataStream.print();

        //5.map包装成value,1
        SingleOutputStreamOperator<Tuple2<String, Integer>> tupleStream = kafkaDataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                if("error".equals(value)){
                    throw new RuntimeException("发生异常!!!");
                }
                return new Tuple2<>(value, 1);
            }
        });
        tupleStream.print();
        //6.按照value进行分组,并且统计value的个数
        SingleOutputStreamOperator<Tuple2<String, Integer>> reduceStream = tupleStream.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
            }
        });


        System.out.println("=====================================================");
        reduceStream.print();
        //7.将数据输出到kafka
        FlinkKafkaProducer<Tuple2<String, Integer>> flinkKafkaProducer = KafkaUtil.getFlinkKafkaProducer(parameterTool, new KafkaSerializationSchema<Tuple2<String, Integer>>() {
            @Override
            public void open(SerializationSchema.InitializationContext context) throws Exception {
                System.out.println("=========正在向KafkaProduce输出数据!!!=============");
            }

            @Override
            public ProducerRecord<byte[], byte[]> serialize(Tuple2<String, Integer> element, @Nullable Long timestamp) {
                String producetopics = parameterTool.get("Producetopice");
                String result = element.toString();
                return new ProducerRecord<byte[], byte[]>(producetopics, result.getBytes(StandardCharsets.UTF_8));
            }
        });
        reduceStream.addSink(flinkKafkaProducer).name("kafkasinktest").uid("kafkasink");

        //任务执行
        env.execute("KafkaToKafkaTest");
    }

注意事项:
这里使用的是本地FSstateBackend,注意你的路径的设置,以hdfs://或者file://为地址标识符,否则Flink的文件系统将无法识别。

三、消费者消费kafka的事务数据

ublic static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties sourceProperties = new Properties();
        sourceProperties.setProperty("bootstrap.servers""*****");
        sourceProperties.setProperty("group.id""****");
        //端到端一致性:消费数据时需要配置isolation.level=read_committed(默认值为read_uncommitted)
        sourceProperties.put("isolation.level""read_committed");

        FlinkKafkaConsumer<String> ConsumerKafka = new FlinkKafkaConsumer<>("*****", new SimpleStringSchema(), sourceProperties);
        ConsumerKafka.setStartFromEarliest();
        DataStreamSource<String> dataStreamSource = env.addSource(ConsumerKafka);

        dataStreamSource.print();

        env.execute();
    }

isolation.level这里设置为read_committed(默认为read_uncommitted) 这里可以看到以你CheckPoint设置的时间,来批量展示kafka生产者的消息。

四、总结与可能出现的问题

以上是flink 实现kafka的精确一次的测试例子,这里还有一点要注意,就是小伙伴们的kafka的配置里面。

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
default.replication.factor=1

这四个参数里面default.replication.factor是你kafka真正每个topic的副本数量,但是在开启事务也就是flink的addsink的时候会默认继承两阶段提交的方式,这里transaction.state.log.replication.factor一定要大于或者等于transaction.state.log.min.isr,否则你的kafka集群不满足事务副本复制的基本属性,会一直不成功,那么你的CheckPoint就会超时过期,从而导致任务的整体失败。

kafka集群第一次有消费者消费消息时会自动创建 __consumer_offsets,它的副本因子受 offsets.topic.replication.factor 参数的约束,默认值为3(注意:该参数的使用限制在0.11.0.0版本发生变化),分区数可以通过 offsets.topic.num.partitions 参数设置,默认值为50,在开启事务性的情况下就会首先会获得一个全局的TransactionCoordinator id和transactional producer并且生成唯一的序列号等 类似于一下的例子来唯一标识当前事务的消息对应的offset,以及标识。

[2022-03-24 21:07:40,022] INFO [TransactionCoordinator id=0] Initialized transactionalId Keyed Reduce -> (Sink: Print to Std. Out, Sink: kafkasinktest)-b0c5e26be6392399cc3c8a38581a81c2-8 with producerId 11101 and producer epoch 8 on partition __transaction_state-18 (kafka.coordinator.transaction.TransactionCoordinator)

当flink任务出现异常的情况下,kafka会把以及提交但是未标记可以消费的数据直接销毁,或者正常的情况下,会正式提交(本质是修改消息的标志位),之后对于消费者在开启isolation.level的时候就可以读取以及标记为可以读取的message。


如果这个文章对你有帮助,不要忘记 「在看」 「点赞」 「收藏」 三连啊喂!

2022年全网首发|大数据专家级技能模型与学习指南(胜天半子篇)
互联网最坏的时代可能真的来了
我在B站读大学,大数据专业
我们在学习Flink的时候,到底在学习什么?
193篇文章暴揍Flink,这个合集你需要关注一下
Flink生产环境TOP难题与优化,阿里巴巴藏经阁YYDS
Flink CDC我吃定了耶稣也留不住他!| Flink CDC线上问题小盘点
我们在学习Spark的时候,到底在学习什么?
在所有Spark模块中,我愿称SparkSQL为最强!
硬刚Hive | 4万字基础调优面试小总结
数据治理方法论和实践小百科全书
标签体系下的用户画像建设小指南
4万字长文 | ClickHouse基础&实践&调优全视角解析
【面试&个人成长】2021年过半,社招和校招的经验之谈
大数据方向另一个十年开启 |《硬刚系列》第一版完结
我写过的关于成长/面试/职场进阶的文章
当我们在学习Hive的时候在学习什么?「硬刚Hive续集」

浏览 32
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报