Flink的sink实战之二：kafka

Posted 程序员欣宸

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Flink的sink实战之二：kafka相关的知识，希望对你有一定的参考价值。

欢迎访问我的GitHub

版本和环境准备

  • 本次实战的环境和版本如下：
    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

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

源码下载

名称 链接 备注
项目主页 https://github.com/zq2599/blog_demos 该项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blog_demos.git 该项目源码的仓库地址，https协议
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址，ssh协议
  • 这个git项目中有多个文件夹，本章的应用在==flinksinkdemo==文件夹下，如下图红框所示：
  • 准备完毕，开始开发；

    准备工作

  • 正式编码前，先去官网查看相关资料了解基本情况：
    1. 地址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html
    2. 我这里用的kafka是2.4.0版本，在官方文档查找对应的库和类，如下图红框所示：

      kafka准备

    3. 创建名为test006的topic，有四个分区，参考命令： 
      ./kafka-topics.sh \\
--create \\
--bootstrap-server 127.0.0.1:9092 \\
--replication-factor 1 \\
--partitions 4 \\
--topic test006
    4. 在控制台消费test006的消息，参考命令： 
      ./kafka-console-consumer.sh \\
--bootstrap-server 127.0.0.1:9092 \\
--topic test006
    5. 此时如果该topic有消息进来，就会在控制台输出；
    6. 接下来开始编码；

      创建工程

    7. 用maven命令创建flink工程： 
      mvn \\
archetype:generate \\
-DarchetypeGroupId=org.apache.flink \\
-DarchetypeArtifactId=flink-quickstart-java \\
-DarchetypeVersion=1.9.2
    8. 根据提示，groupid输入==com.bolingcavalry==，artifactid输入==flinksinkdemo==，即可创建一个maven工程；
    9. 在pom.xml中增加kafka依赖库： 
      <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.9.0</version>
</dependency>
    10. 工程创建完成，开始编写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<String> 

private String topic;

public ProducerStringSerializationSchema(String topic) 
    super();
    this.topic = topic;


@Override
public ProducerRecord<byte[], byte[]> serialize(String element, Long timestamp) 
    return new ProducerRecord<byte[], byte[]>(topic, element.getBytes(StandardCharsets.UTF_8));


2. 创建任务类KafkaStrSink，请注意FlinkKafkaProducer对象的参数，FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次：
```java
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<String> producer = new FlinkKafkaProducer<>(topic,
                new ProducerStringSerializationSchema(topic),
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        //创建一个List，里面有两个Tuple2元素
        List<String> 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");
  1. 使用mvn命令编译构建，在target目录得到文件==flinksinkdemo-1.0-SNAPSHOT.jar==；
  2. 在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制定执行类，如下图：
  3. 提交成功后，如果flink有四个可用slot，任务会立即执行，会在消费kafak消息的终端收到消息，如下图：

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

    发送对象消息的sink

    • 再来尝试如何发送对象类型的消息，这里的对象选择常用的Tuple2对象：
  5. 创建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<Tuple2<String, Integer>> 

private String topic;
private ObjectMapper mapper;

public ObjSerializationSchema(String topic) 
    super();
    this.topic = topic;


@Override
public ProducerRecord<byte[], byte[]> serialize(Tuple2<String, Integer> 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<byte[], byte[]>(topic, b);



2. 创建flink任务类：
```java
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<Tuple2<String, Integer>> producer = new FlinkKafkaProducer<>(topic,
                new ObjSerializationSchema(topic),
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        //创建一个List，里面有两个Tuple2元素
        List<Tuple2<String, Integer>> 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");
  1. 像前一个任务那样编译构建，把jar提交到flink，并指定执行类是==com.bolingcavalry.addsink.KafkaObjSink==；
  2. 消费kafka消息的控制台输出如下：
  3. 在web页面可见执行情况如下：
    • 至此，flink将计算结果作为kafka消息发送出去的实战就完成了，希望能给您提供参考，接下来的章节，我们会继续体验官方提供的sink能力；

欢迎关注51CTO博客：程序员欣宸

以上是关于Flink的sink实战之二：kafka的主要内容，如果未能解决你的问题，请参考以下文章

Flink的sink实战之四：自定义

Flink的sink实战之一：初探

Flink流式计算从入门到实战 三

flink使用kafka作为source和sink

Apache Flink：使用Apache Kafka作为Sink的简单demo(数据结果存放地方)

Flink 实战系列Flink 同步 Kafka 数据到 HDFS parquet 格式存储 snappy 压缩