本文是《Flink的sink實戰》系列的第二篇，《 》對sink有了基本的瞭解，本章來體驗將數據sink到kafka的操作；
版本和環境準備
本次實戰的環境和版本如下：
- 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
請確保上述環境和服務已經就緒；
源碼下載
如果您不想寫代碼，整個系列的源碼可在GitHub下載到，地址和鏈接信息如下表所示：
這個git項目中有多個文件夾，本章的應用在flinksinkdemo文件夾下，如下圖紅框所示：
準備工作
正式編碼前，先去官網查看相關資料瞭解基本情況：
kafka準備
- 創建名為test006的topic，有四個分區，參考命令：
<code>./kafka-topics.sh \\
--create \\
--bootstrap-server 127.0.0.1:9092 \\
--replication-factor 1 \\
--partitions 4 \\
--topic test006/<code>
- 在控制檯消費test006的消息，參考命令：
<code>./kafka-console-consumer.sh \\
--bootstrap-server 127.0.0.1:9092 \\
--topic test006/<code>
- 此時如果該topic有消息進來，就會在控制檯輸出；
- 接下來開始編碼；
創建工程
- 用maven命令創建flink工程：
<code>mvn \\
archetype:generate \\
-DarchetypeGroupId=org.apache.flink \\
-DarchetypeArtifactId=flink-quickstart-java \\
-DarchetypeVersion=1.9.2/<code>
- 根據提示，groupid輸入com.bolingcavalry，artifactid輸入flinksinkdemo，即可創建一個maven工程；
- 在pom.xml中增加kafka依賴庫：
<code><dependency>
<groupid>org.apache.flink/<groupid>
<artifactid>flink-connector-kafka_2.11/<artifactid>
<version>1.9.0/<version>
/<dependency>/<code>
- 工程創建完成，開始編寫flink任務的代碼；
發送字符串消息的sink
先嚐試發送字符串類型的消息：
- 創建KafkaSerializationSchema接口的實現類，後面這個類要作為創建sink對象的參數使用：
<code>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> serialize(String element, Long timestamp) {
return new ProducerRecord<byte>(topic, element.getBytes(StandardCharsets.UTF_8));
}
}/<byte>/<byte>/<string>/<code>
- 創建任務類KafkaStrSink，請注意FlinkKafkaProducer對象的參數，FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示嚴格一次：
<code>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");
}
}/<string>/<string>/<code>
- 提交成功後，如果flink有四個可用slot，任務會立即執行，會在消費kafak消息的終端收到消息，如下圖：
- 任務執行情況如下圖：
發送對象消息的sink
再來嘗試如何發送對象類型的消息，這裡的對象選擇常用的Tuple2對象：
- 創建KafkaSerializationSchema接口的實現類，該類後面要用作sink對象的入參，請注意代碼中捕獲異常的那段註釋： 生產環境慎用printStackTrace()!!!
<code>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>> {
private String topic;
private ObjectMapper mapper;
public ObjSerializationSchema(String topic) {
super();
this.topic = topic;
}
@Override
public ProducerRecord<byte> serialize(Tuple2<string> 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>(topic, b);
}
}/<byte>/<string>/<byte>/<tuple2>/<code>
- 創建flink任務類：
<code>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>> producer = new FlinkKafkaProducer<>(topic,
new ObjSerializationSchema(topic),
properties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
//創建一個List，裡面有兩個Tuple2元素
List<tuple2>> 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");
}
}/<tuple2>/<tuple2>/<code>
- 像前一個任務那樣編譯構建，把jar提交到flink，並指定執行類是com.bolingcavalry.addsink.KafkaObjSink；
- 消費kafka消息的控制檯輸出如下：
- 在web頁面可見執行情況如下：
至此，flink將計算結果作為kafka消息發送出去的實戰就完成了，希望能給您提供參考，接下來的章節，我們會繼續體驗官方提供的sink能力；
分享到:
閱讀更多 程序員欣宸 的文章
