kafka maven 依賴_Flink的sink實戰之二：kafka

歡迎通路我的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

請確定上述環境和服務已經就緒；

源碼下載下傳

如果您不想寫代碼，整個系列的源碼可在GitHub下載下傳到，位址和連結資訊如下表所示(https://github.com/zq2599/blog_demos)：

這個git項目中有多個檔案夾，本章的應用在flinksinkdemo檔案夾下，如下圖紅框所示：

kafka maven 依賴_Flink的sink實戰之二：kafka

準備完畢，開始開發；

準備工作

正式編碼前，先去官網檢視相關資料了解基本情況：

位址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html
我這裡用的kafka是2.4.0版本，在官方文檔查找對應的庫和類，如下圖紅框所示：

kafka maven 依賴_Flink的sink實戰之二：kafka

kafka準備
建立名為test006的topic，有四個分區，參考指令： ```shell ./kafka-topics.sh

--create

--bootstrap-server 127.0.0.1:9092

--replication-factor 1

--partitions 4

--topic test006

2. 在控制台消費test006的消息，參考指令：
```shell
./kafka-console-consumer.sh 
--bootstrap-server 127.0.0.1:9092 
--topic test006

此時如果該topic有消息進來，就會在控制台輸出；
接下來開始編碼；
建立工程
用maven指令建立flink工程： ```shell mvn

archetype:generate

-DarchetypeGroupId=org.apache.flink

-DarchetypeArtifactId=flink-quickstart-java

-DarchetypeVersion=1.9.2

2. 根據提示，groupid輸入<font color="blue">com.bolingcavalry</font>，artifactid輸入<font color="blue">flinksinkdemo</font>，即可建立一個maven工程；
3. 在pom.xml中增加kafka依賴庫：
```xml
<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 {

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，并制定執行類，如下圖：

kafka maven 依賴_Flink的sink實戰之二：kafka
送出成功後，如果flink有四個可用slot，任務會立即執行，會在消費kafak消息的終端收到消息，如下圖：

kafka maven 依賴_Flink的sink實戰之二：kafka
任務執行情況如下圖：

kafka maven 依賴_Flink的sink實戰之二：kafka

發送對象消息的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任務類：

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消息的控制台輸出如下：

kafka maven 依賴_Flink的sink實戰之二：kafka
在web頁面可見執行情況如下：

kafka maven 依賴_Flink的sink實戰之二：kafka
至此，flink将計算結果作為kafka消息發送出去的實戰就完成了，希望能給您提供參考，接下來的章節，我們會繼續體驗官方提供的sink能力；

歡迎關注公衆号：程式員欣宸

微信搜尋「程式員欣宸」，我是欣宸，期待與您一同暢遊Java世界... https://github.com/zq2599/blog_demos

kafka maven 依賴_Flink的sink實戰之二：kafka

歡迎通路我的GitHub

本篇概覽

全系列連結

版本和環境準備

源碼下載下傳

準備工作

kafka準備

建立工程

發送字元串消息的sink

發送對象消息的sink

歡迎關注公衆号：程式員欣宸

繼續閱讀

batchplot3.5.9如何使用_Flink 1.9 實戰：使用 SQL 讀取 Kafka 并寫入 MySQL

kafka maven 依賴_Spring Boot 2.x 整合消息中間件 Kafka

kafka maven 依賴_SpringMVC整合Kafka實戰1.SpringMVC整合生産者(Producer)2.SpringMVC整合消費者(Consumer)

kafka maven 依賴_Flink的sink實戰之二：kafka版本和環境準備源碼下載下傳準備工作kafka準備建立工程發送字元串消息的sink發送對象消息的sink歡迎關注公衆号：程式員欣宸