乐趣区

关于云计算:Flink的sink实战之二kafka

作者：

jiezi

在

欢送拜访我的GitHub

https://github.com/zq2599/blog_demos

内容：所有原创文章分类汇总及配套源码，波及Java、Docker、Kubernetes、DevOPS等；

本篇概览

本文是《Flink的sink实战》系列的第二篇，前文《Flink的sink实战之一：初探》对sink有了根本的理解，本章来体验将数据sink到kafka的操作；

全系列链接

  1. 《Flink的sink实战之一：初探》
  2. 《Flink的sink实战之二：kafka》
  3. 《Flink的sink实战之三：cassandra3》
  4. 《Flink的sink实战之四：自定义》

版本和环境筹备

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

  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

<font color=”red”>请确保上述环境和服务曾经就绪；</font>

源码下载

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

名称 链接 备注
我的项目主页 https://github.com/zq2599/blo… 该我的项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blo… 该我的项目源码的仓库地址，https协定
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该我的项目源码的仓库地址，ssh协定

这个git我的项目中有多个文件夹，本章的利用在<font color=”blue”>flinksinkdemo</font>文件夹下，如下图红框所示：

筹备结束，开始开发；

筹备工作

正式编码前，先去官网查看相干材料理解根本状况：

  1. 地址：https://ci.apache.org/project…
  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
  1. 在控制台生产test006的音讯，参考命令：
./kafka-console-consumer.sh \
--bootstrap-server 127.0.0.1:9092 \
--topic test006
  1. 此时如果该topic有音讯进来，就会在控制台输入；
  2. 接下来开始编码；

创立工程

  1. 用maven命令创立flink工程：
mvn \
archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.9.2
  1. 依据提醒，groupid输出<font color=”blue”>com.bolingcavalry</font>，artifactid输出<font color=”blue”>flinksinkdemo</font>，即可创立一个maven工程；
  2. 在pom.xml中减少kafka依赖库：
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.11</artifactId>
  <version>1.9.0</version>
</dependency>
  1. 工程创立实现，开始编写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));
    }
}
  1. 创立工作类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<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目录失去文件<font color=”blue”>flinksinkdemo-1.0-SNAPSHOT.jar</font>；
  2. 在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制订执行类，如下图：

  1. 提交胜利后，如果flink有四个可用slot，工作会立刻执行，会在生产kafak音讯的终端收到音讯，如下图：

  1. 工作执行状况如下图：

发送对象音讯的sink

再来尝试如何发送对象类型的音讯，这里的对象抉择罕用的Tuple2对象：

  1. 创立KafkaSerializationSchema接口的实现类，该类前面要用作sink对象的入参，请留神代码中捕捉异样的那段正文：<font color=”red”>生产环境慎用printStackTrace()!!!</font>
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);
    }
}
  1. 创立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<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，并指定执行类是<font color=”blue”>com.bolingcavalry.addsink.KafkaObjSink</font>；
  2. 生产kafka音讯的控制台输入如下：

  1. 在web页面可见执行状况如下：


至此，flink将计算结果作为kafka音讯发送进来的实战就实现了，心愿能给您提供参考，接下来的章节，咱们会持续体验官网提供的sink能力；

欢送关注公众号：程序员欣宸

微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界…
https://github.com/zq2599/blog_demos

更多文章