如何进行数据sink到kafka的操作

这篇文章给大家介绍如何进行数据sink到kafka的操作，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。

下面来体验将数据sink到kafka的操作。

版本和环境准备

本次实战的环境和版本如下：

1. JDK：1.8.0_211

2. Flink：1.9.2

3. Maven：3.6.0

4. 操作系统：macOS Catalina 10.15.3 （MacBook Pro 13-inch, 2018）

5. IDEA：2018.3.5 (Ultimate Edition)

6. Kafka：2.4.0

7. Zookeeper：3.5.5

请确保上述环境和服务已经就绪；

源码下载

如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog_demos)：

这个git项目中有多个文件夹，本章的应用在flinksinkdemo文件夹下，如下图红框所示： 准备完毕，开始开发；

准备工作

正式编码前，先去官网查看相关资料了解基本情况：

1. 地址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html

2. 我这里用的kafka是2.4.0版本，在官方文档查找对应的库和类，如下图红框所示：

kafka准备

1. 创建名为test006的topic，有四个分区，参考命令：

./kafka-topics.sh \--create \--bootstrap-server 127.0.0.1:9092 \--replication-factor 1 \--partitions 4 \--topic test006

2. 在控制台消费test006的消息，参考命令：

./kafka-console-consumer.sh \--bootstrap-server 127.0.0.1:9092 \--topic test006

3. 此时如果该topic有消息进来，就会在控制台输出；

4. 接下来开始编码；

创建工程

1. 用maven命令创建flink工程：

mvn \archetype:generate \-DarchetypeGroupId=org.apache.flink \-DarchetypeArtifactId=flink-quickstart-java \-DarchetypeVersion=1.9.2

2. 根据提示，groupid输入com.bolingcavalry，artifactid输入flinksinkdemo，即可创建一个maven工程；

3. 在pom.xml中增加kafka依赖库：

  org.apache.flink  flink-connector-kafka_2.11  1.9.0

4. 工程创建完成，开始编写flink任务的代码；

发送字符串消息的sink

先尝试发送字符串类型的消息：

1. 创建KafkaSerializationSchema接口的实现类，后面这个类要作为创建sink对象的参数使用：

package com.bolingcavalry.addsink;import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;import org.apache.kafka.clients.producer.ProducerRecord;import java.nio.charset.StandardCharsets;public class ProducerStringSerializationSchema implements KafkaSerializationSchema {    private String topic;    public ProducerStringSerializationSchema(String topic) {        super();        this.topic = topic;    }    @Override    public ProducerRecord serialize(String element, Long timestamp) {        return new ProducerRecord(topic, element.getBytes(StandardCharsets.UTF_8));    }}

2. 创建任务类KafkaStrSink，请注意FlinkKafkaProducer对象的参数，FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次：

package com.bolingcavalry.addsink;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import java.util.ArrayList;import java.util.List;import java.util.Properties;public class KafkaStrSink {    public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        //并行度为1        env.setParallelism(1);        Properties properties = new Properties();        properties.setProperty("bootstrap.servers", "192.168.50.43:9092");        String topic = "test006";        FlinkKafkaProducer producer = new FlinkKafkaProducer<>(topic,                new ProducerStringSerializationSchema(topic),                properties,                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);        //创建一个List，里面有两个Tuple2元素        List list = new ArrayList<>();        list.add("aaa");        list.add("bbb");        list.add("ccc");        list.add("ddd");        list.add("eee");        list.add("fff");        list.add("aaa");        //统计每个单词的数量        env.fromCollection(list)           .addSink(producer)           .setParallelism(4);        env.execute("sink demo : kafka str");    }}

3. 使用mvn命令编译构建，在target目录得到文件flinksinkdemo-1.0-SNAPSHOT.jar；

4. 在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制定执行类，如下图：

5. 提交成功后，如果flink有四个可用slot，任务会立即执行，会在消费kafak消息的终端收到消息，如下图：

6. 任务执行情况如下图：

发送对象消息的sink

再来尝试如何发送对象类型的消息，这里的对象选择常用的Tuple2对象：

1. 创建KafkaSerializationSchema接口的实现类，该类后面要用作sink对象的入参，请注意代码中捕获异常的那段注释：生产环境慎用printStackTrace()!!!

package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;import org.apache.kafka.clients.producer.ProducerRecord;import javax.annotation.Nullable;public class ObjSerializationSchema implements KafkaSerializationSchema> {    private String topic;    private ObjectMapper mapper;    public ObjSerializationSchema(String topic) {        super();        this.topic = topic;    }    @Override    public ProducerRecord serialize(Tuple2 stringIntegerTuple2, @Nullable Long timestamp) {        byte[] b = null;        if (mapper == null) {            mapper = new ObjectMapper();        }        try {            b= mapper.writeValueAsBytes(stringIntegerTuple2);        } catch (JsonProcessingException e) {            // 注意，在生产环境这是个非常危险的操作，            // 过多的错误打印会严重影响系统性能，请根据生产环境情况做调整            e.printStackTrace();        }        return new ProducerRecord(topic, b);    }}

2. 创建flink任务类：

package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import java.util.ArrayList;import java.util.List;import java.util.Properties;public class KafkaObjSink {    public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        //并行度为1        env.setParallelism(1);        Properties properties = new Properties();        //kafka的broker地址        properties.setProperty("bootstrap.servers", "192.168.50.43:9092");        String topic = "test006";        FlinkKafkaProducer> producer = new FlinkKafkaProducer<>(topic,                new ObjSerializationSchema(topic),                properties,                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);        //创建一个List，里面有两个Tuple2元素        List> list = new ArrayList<>();        list.add(new Tuple2("aaa", 1));        list.add(new Tuple2("bbb", 1));        list.add(new Tuple2("ccc", 1));        list.add(new Tuple2("ddd", 1));        list.add(new Tuple2("eee", 1));        list.add(new Tuple2("fff", 1));        list.add(new Tuple2("aaa", 1));        //统计每个单词的数量        env.fromCollection(list)            .keyBy(0)            .sum(1)            .addSink(producer)            .setParallelism(4);                env.execute("sink demo : kafka obj");    }}

3. 像前一个任务那样编译构建，把jar提交到flink，并指定执行类是com.bolingcavalry.addsink.KafkaObjSink；

4. 消费kafka消息的控制台输出如下：

5. 在web页面可见执行情况如下： 至此，flink将计算结果作为kafka消息发送出去的实战就完成了，希望能给您提供参考，接下来的章节，我们会继续体验官方提供的sink能力；

关于如何进行数据sink到kafka的操作就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。