如何進(jìn)行數(shù)據(jù)sink到kafka的操作

這篇文章給大家介紹如何進(jìn)行數(shù)據(jù)sink到kafka的操作，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

下面來體驗將數(shù)據(jù)sink到kafka的操作。

版本和環(huán)境準(zhǔn)備

本次實戰(zhàn)的環(huán)境和版本如下：

1. JDK：1.8.0_211

2. Flink：1.9.2

3. Maven：3.6.0

4. 操作系統(tǒng)：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

<font color="red">請確保上述環(huán)境和服務(wù)已經(jīng)就緒；</font>

源碼下載

如果您不想寫代碼，整個系列的源碼可在GitHub下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：

這個git項目中有多個文件夾，本章的應(yīng)用在<font color="blue">flinksinkdemo</font>文件夾下，如下圖紅框所示： 準(zhǔn)備完畢，開始開發(fā)；

準(zhǔn)備工作

正式編碼前，先去官網(wǎng)查看相關(guān)資料了解基本情況：

1. 地址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html

2. 我這里用的kafka是2.4.0版本，在官方文檔查找對應(yīng)的庫和類，如下圖紅框所示：

kafka準(zhǔn)備

1. 創(chuàng)建名為test006的topic，有四個分區(qū)，參考命令：

./kafka-topics.sh \
--create \
--bootstrap-server 127.0.0.1:9092 \
--replication-factor 1 \
--partitions 4 \
--topic test006

2. 在控制臺消費(fèi)test006的消息，參考命令：

./kafka-console-consumer.sh \
--bootstrap-server 127.0.0.1:9092 \
--topic test006

3. 此時如果該topic有消息進(jìn)來，就會在控制臺輸出；

4. 接下來開始編碼；

創(chuàng)建工程

1. 用maven命令創(chuàng)建flink工程：

mvn \
archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.9.2

2. 根據(jù)提示，groupid輸入<font color="blue">com.bolingcavalry</font>，artifactid輸入<font color="blue">flinksinkdemo</font>，即可創(chuàng)建一個maven工程；

3. 在pom.xml中增加kafka依賴庫：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.11</artifactId>
  <version>1.9.0</version>
</dependency>

4. 工程創(chuàng)建完成，開始編寫flink任務(wù)的代碼；

發(fā)送字符串消息的sink

先嘗試發(fā)送字符串類型的消息：

1. 創(chuàng)建KafkaSerializationSchema接口的實現(xiàn)類，后面這個類要作為創(chuàng)建sink對象的參數(shù)使用：

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. 創(chuàng)建任務(wù)類KafkaStrSink，請注意FlinkKafkaProducer對象的參數(shù)，F(xiàn)linkKafkaProducer.Semantic.EXACTLY_ONCE表示嚴(yán)格一次：

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

        //創(chuàng)建一個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");

        //統(tǒng)計每個單詞的數(shù)量
        env.fromCollection(list)
           .addSink(producer)
           .setParallelism(4);

        env.execute("sink demo : kafka str");
    }
}

3. 使用mvn命令編譯構(gòu)建，在target目錄得到文件<font color="blue">flinksinkdemo-1.0-SNAPSHOT.jar</font>；

4. 在flink的web頁面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制定執(zhí)行類，如下圖：

5. 提交成功后，如果flink有四個可用slot，任務(wù)會立即執(zhí)行，會在消費(fèi)kafak消息的終端收到消息，如下圖：

6. 任務(wù)執(zhí)行情況如下圖：

發(fā)送對象消息的sink

再來嘗試如何發(fā)送對象類型的消息，這里的對象選擇常用的Tuple2對象：

1. 創(chuàng)建KafkaSerializationSchema接口的實現(xiàn)類，該類后面要用作sink對象的入?yún)ⅲ堊⒁獯a中捕獲異常的那段注釋：<font color="red">生產(chǎn)環(huán)境慎用printStackTrace()!!!</font>

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) {
            // 注意，在生產(chǎn)環(huán)境這是個非常危險的操作，
            // 過多的錯誤打印會嚴(yán)重影響系統(tǒng)性能，請根據(jù)生產(chǎn)環(huán)境情況做調(diào)整
            e.printStackTrace();
        }
        return new ProducerRecord<byte[], byte[]>(topic, b);
    }
}

2. 創(chuàng)建flink任務(wù)類：

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

        //創(chuàng)建一個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));

        //統(tǒng)計每個單詞的數(shù)量
        env.fromCollection(list)
            .keyBy(0)
            .sum(1)
            .addSink(producer)
            .setParallelism(4);
        
        env.execute("sink demo : kafka obj");
    }
}

3. 像前一個任務(wù)那樣編譯構(gòu)建，把jar提交到flink，并指定執(zhí)行類是<font color="blue">com.bolingcavalry.addsink.KafkaObjSink</font>；

4. 消費(fèi)kafka消息的控制臺輸出如下：

5. 在web頁面可見執(zhí)行情況如下： 至此，flink將計算結(jié)果作為kafka消息發(fā)送出去的實戰(zhàn)就完成了，希望能給您提供參考，接下來的章節(jié)，我們會繼續(xù)體驗官方提供的sink能力；

關(guān)于如何進(jìn)行數(shù)據(jù)sink到kafka的操作就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學(xué)到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。