本文是《Flink的sink實戰》系列的第二篇，《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檔案夾下，如下圖紅框所示：
準備工作
正式編碼前，先去官網檢視相關資料了解基本情況：
位址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html我這裡用的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輸入com.bolingcavalry，artifactid輸入flinksinkdemo，即可建立一個maven工程；
- 在pom.xml中增加kafka依賴庫：
org.apache.flink flink-connector-kafka_2.11 1.9.0
- 工程建立完成，開始編寫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 { private String topic; public ProducerStringSerializationSchema(String topic) { super(); this.topic = topic; } @Override public ProducerRecord serialize(String element, Long timestamp) { return new ProducerRecord(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 producer = new FlinkKafkaProducer<>(topic, new ProducerStringSerializationSchema(topic), properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE); //建立一個List，裡面有兩個Tuple2元素 List 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> { private String topic; private ObjectMapper mapper; public ObjSerializationSchema(String topic) { super(); this.topic = topic; } @Override public ProducerRecord serialize(Tuple2 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(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> producer = new FlinkKafkaProducer<>(topic, new ObjSerializationSchema(topic), properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE); //建立一個List，裡面有兩個Tuple2元素 List> 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能力；
歡迎關注公衆号：程式員欣宸