Flink的sink实战之二（kafka） _FLINK

亦余心之所善兮，虽九死其犹未悔。这篇文章主要讲述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

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

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

名称	链接	备注
项目主页	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");

使用mvn命令编译构建，在target目录得到文件==flinksinkdemo-1.0-SNAPSHOT.jar==；
在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制定执行类，如下图：

文章图片
提交成功后，如果flink有四个可用slot，任务会立即执行，会在消费kafak消息的终端收到消息，如下图：

文章图片
任务执行情况如下图：

文章图片

发送对象消息的sink
- 再来尝试如何发送对象类型的消息，这里的对象选择常用的Tuple2对象：
创建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);

【Flink的sink实战之二（kafka）】

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");

像前一个任务那样编译构建，把jar提交到flink，并指定执行类是==com.bolingcavalry.addsink.KafkaObjSink==；
消费kafka消息的控制台输出如下：

文章图片
在web页面可见执行情况如下：

文章图片
- 至此，flink将计算结果作为kafka消息发送出去的实战就完成了，希望能给您提供参考，接下来的章节，我们会继续体验官方提供的sink能力；