Flink Flink数据写入Kafka

一、环境准备

官网地址

flink官方集成了通用的 Kafka 连接器，使用时需要根据生产环境的版本引入相应的依赖

    <properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><flink.version>1.14.6</flink.version><spark.version>2.4.3</spark.version><hadoop.version>2.8.5</hadoop.version><hbase.version>1.4.9</hbase.version><hive.version>2.3.5</hive.version><java.version>1.8</java.version><scala.version>2.11.8</scala.version><mysql.version>8.0.22</mysql.version><scala.binary.version>2.11</scala.binary.version><maven.compiler.source>${java.version}</maven.compiler.source><maven.compiler.target>${java.version}</maven.compiler.target></properties>

        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.12</artifactId><version>${flink.version}</version></dependency>

二、KafkaSink介绍

在这里插入图片描述

三、正确理解序列化器

什么叫序列化和反序列化？
1.序列化：把对象转换为字节序列的过程称为对象的序列化.
2.反序列化：把字节序列恢复为对象的过程称为对象的反序列化.

序列化器的作用是将flink数据转换成 kafka的ProducerRecord
在这里插入图片描述
使用预定义的序列化器
将 DataStream 数据转换为 Kafka消息中的value，key为默认值null，timestamp为默认值

        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()// 指定 kafka 的地址和端口.setBootstrapServers("localhost:9092")// 指定序列化器：指定Topic名称、具体的序列化(产生方需要序列化，接收方需要反序列化).setRecordSerializer(KafkaRecordSerializationSchema.<String>builder().setTopic("testtopic01")// 指定value的序列化器.setValueSerializationSchema(new SimpleStringSchema()).build())

源码解析

    public <T extends IN> KafkaRecordSerializationSchemaBuilder<T> setValueSerializationSchema(SerializationSchema<T> valueSerializationSchema) {this.checkValueSerializerNotSet();KafkaRecordSerializationSchemaBuilder<T> self = this.self();self.valueSerializationSchema = (SerializationSchema)Preconditions.checkNotNull(valueSerializationSchema);return self;}

使用自定义的序列化器

KafkaSink<String> kafkaSink = KafkaSink.<String>builder()// TODO 必填项：配置 kafka 的地址和端口.setBootstrapServers("localhost:9092")// TODO 必填项：配置消息序列化器信息 Topic名称、消息序列化器类型.setRecordSerializer(new KafkaRecordSerializationSchema<String>() {...............}).build();

四、容错保证级别

KafkaSink 总共支持三种不同的语义保证（DeliveryGuarantee）

DeliveryGuarantee.NONE 不提供任何保证
消息有可能会因 Kafka broker 的原因发生丢失或因 Flink 的故障发生重复

DeliveryGuarantee.AT_LEAST_ONCE 至少一次
sink 在 checkpoint 时会等待 Kafka 缓冲区中的数据全部被 Kafka producer 确认。
消息不会因 Kafka broker 端发生的事件而丢失，但可能会在 Flink 重启时重复，因为 Flink 会重新处理旧数据。

DeliveryGuarantee.EXACTLY_ONCE 精确一次
该模式下，Kafka sink 会将所有数据通过在 checkpoint 时提交的事务写入。
因此，如果 consumer 只读取已提交的数据（参见 Kafka consumer 配置 isolation.level），在 Flink 发生重启时不会发生数据重复。
然而这会使数据在 checkpoint 完成时才会可见，因此请按需调整 checkpoint 的间隔。
请确认事务 ID 的前缀（transactionIdPrefix）对不同的应用是唯一的，以保证不同作业的事务不会互相影响！此外，强烈建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间，否则 Kafka 对未提交事务的过期处理会导致数据丢失。

五、案例—Flink将Socket数据写入Kafka（精准一次）

注意：如果要使用精准一次写入 Kafka，需要满足以下条件，缺一不可
1、开启 checkpoint
2、设置事务前缀
3、设置事务超时时间： checkpoint 间隔 < 事务超时时间 < max 的 15 分钟

package com.flink.DataStream.Sink;import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Properties;public class flinkSinkKafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();streamExecutionEnvironment.setParallelism(1);// 如果是精准一次，必须开启 checkpointstreamExecutionEnvironment.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);DataStreamSource<String> streamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);/*** TODO Kafka Sink* TODO 注意：如果要使用 精准一次 写入 Kafka，需要满足以下条件，缺一不可* 1、开启 checkpoint* 2、设置事务前缀* 3、设置事务超时时间： checkpoint 间隔 < 事务超时时间 < max 的 15 分钟*/Properties properties=new Properties();properties.put("transaction.timeout.ms",10 * 60 * 1000 + "");KafkaSink<String> kafkaSink = KafkaSink.<String>builder()// 指定 kafka 的地址和端口.setBootstrapServers("localhost:9092")// 指定序列化器：指定Topic名称、具体的序列化(产生方需要序列化，接收方需要反序列化).setRecordSerializer(KafkaRecordSerializationSchema.<String>builder().setTopic("testtopic01")// 指定value的序列化器.setValueSerializationSchema(new SimpleStringSchema()).build())// 写到 kafka 的一致性级别： 精准一次、至少一次.setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)// 如果是精准一次，必须设置 事务的前缀.setTransactionalIdPrefix("flinkkafkasink-")// 如果是精准一次，必须设置 事务超时时间: 大于 checkpoint间隔，小于 max 15 分钟.setKafkaProducerConfig(properties)//.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "").build();streamSource.sinkTo(kafkaSink);streamExecutionEnvironment.execute();}
}

理解ProduceerConfig配置源码
在这里插入图片描述

六、启动Zookeeper、Kafka

#启动zookeeper
${ZK_HOME}/bin/zkServer.sh start
#查看zookeeper状态
${ZK_HOME}/bin/zkServer.sh status
#启动kafka
${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties
#查看topic
${KAFKA_HOME}/bin/kafka-topics.sh --list --zookeeper localhost:2181
#创建topic
${KAFKA_HOME}/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic testtopic02 --partitions 2 --replication-factor 1
#删除topic
${KAFKA_HOME}/bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic testtopic02
#生产消息
${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic01
#消费消息
${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic01 --from-beginning