乐趣区

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

欢送拜访我的 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

退出移动版