乐趣区

关于云计算: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

退出移动版