欢送拜访我的GitHub
https://github.com/zq2599/blog_demos
内容：所有原创文章分类汇总及配套源码，波及Java、Docker、Kubernetes、DevOPS等；
本篇概览
本文是《Flink的sink实战》系列的第二篇，前文《Flink的sink实战之一：初探》对sink有了根本的理解，本章来体验将数据sink到kafka的操作；
全系列链接
- 《Flink的sink实战之一：初探》
- 《Flink的sink实战之二：kafka》
- 《Flink的sink实战之三：cassandra3》
- 《Flink的sink实战之四：自定义》
版本和环境筹备
本次实战的环境和版本如下：
- JDK：1.8.0_211
- Flink：1.9.2
- Maven：3.6.0
- 操作系统：macOS Catalina 10.15.3 （MacBook Pro 13-inch, 2018）
- IDEA：2018.3.5 (Ultimate Edition)
- Kafka：2.4.0
- 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>文件夹下，如下图红框所示：
筹备结束，开始开发；
筹备工作
正式编码前，先去官网查看相干材料理解根本状况：
- 地址：https://ci.apache.org/project...
- 我这里用的kafka是2.4.0版本，在官网文档查找对应的库和类，如下图红框所示：
kafka筹备
- 创立名为test006的topic，有四个分区，参考命令：
./kafka-topics.sh \--create \--bootstrap-server 127.0.0.1:9092 \--replication-factor 1 \--partitions 4 \--topic test006- 在控制台生产test006的音讯，参考命令：
./kafka-console-consumer.sh \--bootstrap-server 127.0.0.1:9092 \--topic test006- 此时如果该topic有音讯进来，就会在控制台输入；
- 接下来开始编码；
创立工程
- 用maven命令创立flink工程：
mvn \archetype:generate \-DarchetypeGroupId=org.apache.flink \-DarchetypeArtifactId=flink-quickstart-java \-DarchetypeVersion=1.9.2- 依据提醒，groupid输出<font color="blue">com.bolingcavalry</font>，artifactid输出<font color="blue">flinksinkdemo</font>，即可创立一个maven工程；
- 在pom.xml中减少kafka依赖库：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>1.9.0</version></dependency>- 工程创立实现，开始编写flink工作的代码；
发送字符串音讯的sink
先尝试发送字符串类型的音讯：
- 创立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)); }}- 创立工作类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"); }}- 应用mvn命令编译构建，在target目录失去文件<font color="blue">flinksinkdemo-1.0-SNAPSHOT.jar</font>；
- 在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制订执行类，如下图：
- 提交胜利后，如果flink有四个可用slot，工作会立刻执行，会在生产kafak音讯的终端收到音讯，如下图：
- 工作执行状况如下图：
发送对象音讯的sink
再来尝试如何发送对象类型的音讯，这里的对象抉择罕用的Tuple2对象：
- 创立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); }}- 创立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"); }}- 像前一个工作那样编译构建，把jar提交到flink，并指定执行类是<font color="blue">com.bolingcavalry.addsink.KafkaObjSink</font>；
- 生产kafka音讯的控制台输入如下：
- 在web页面可见执行状况如下：
至此，flink将计算结果作为kafka音讯发送进来的实战就实现了，心愿能给您提供参考，接下来的章节，咱们会持续体验官网提供的sink能力；
欢送关注公众号：程序员欣宸
微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界...
https://github.com/zq2599/blog_demos