Flink的sink實戰之二：kafka

歡迎訪問我的GitHub

這里分類和匯總了欣宸的全部原創(含配套源碼)：https://github.com/zq2599/blog_demos

版本和環境準備

本次實戰的環境和版本如下：

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下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：

這個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能力；

歡迎關注華為云博客：程序員欣宸

學習路上，你不孤單，欣宸原創一路相伴…

Flink Kafka

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。