《從0到1學(xué)習(xí)Flink》—— Flink 寫(xiě)入數(shù)據(jù)到 Kafka

image

前言

之前文章《從0到1學(xué)習(xí)Flink》—— Flink 寫(xiě)入數(shù)據(jù)到 ElasticSearch 寫(xiě)了如何將 Kafka 中的數(shù)據(jù)存儲(chǔ)到 ElasticSearch 中孽糖，里面其實(shí)就已經(jīng)用到了 Flink 自帶的 Kafka source connector（FlinkKafkaConsumer）皮官。存入到 ES 只是其中一種情況，那么如果我們有多個(gè)地方需要這份通過(guò) Flink 轉(zhuǎn)換后的數(shù)據(jù)愿吹，是不是又要我們繼續(xù)寫(xiě)個(gè) sink 的插件呢？確實(shí)惜姐，所以 Flink 里面就默認(rèn)支持了不少 sink犁跪，比如也支持 Kafka sink connector（FlinkKafkaProducer），那么這篇文章我們就講講如何將數(shù)據(jù)寫(xiě)入到 Kafka歹袁。

image

準(zhǔn)備

添加依賴(lài)

Flink 里面支持 Kafka 0.8坷衍、0.9、0.10条舔、0.11 枫耳，以后有時(shí)間可以分析下源碼的實(shí)現(xiàn)。

image

這里我們需要安裝下 Kafka孟抗，請(qǐng)對(duì)應(yīng)添加對(duì)應(yīng)的 Flink Kafka connector 依賴(lài)的版本迁杨，這里我們使用的是 0.11 版本：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

Kafka 安裝

這里就不寫(xiě)這塊內(nèi)容了，可以參考我以前的文章 Kafka 安裝及快速入門(mén)凄硼。

這里我們演示把其他 Kafka 集群中 topic 數(shù)據(jù)原樣寫(xiě)入到自己本地起的 Kafka 中去铅协。

配置文件

kafka.brokers=xxx:9092,xxx:9092,xxx:9092
kafka.group.id=metrics-group-test
kafka.zookeeper.connect=xxx:2181
metrics.topic=xxx
stream.parallelism=5
kafka.sink.brokers=localhost:9092
kafka.sink.topic=metric-test
stream.checkpoint.interval=1000
stream.checkpoint.enable=false
stream.sink.parallelism=5

目前我們先看下本地 Kafka 是否有這個(gè) metric-test topic 呢？需要執(zhí)行下這個(gè)命令：

bin/kafka-topics.sh --list --zookeeper localhost:2181

image

可以看到本地的 Kafka 是沒(méi)有任何 topic 的摊沉，如果等下我們的程序運(yùn)行起來(lái)后警医，再次執(zhí)行這個(gè)命令出現(xiàn) metric-test topic，那么證明我的程序確實(shí)起作用了坯钦，已經(jīng)將其他集群的 Kafka 數(shù)據(jù)寫(xiě)入到本地 Kafka 了预皇。

程序代碼

Main.java

public class Main {
    public static void main(String[] args) throws Exception{
        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
        StreamExecutionEnvironment env = ExecutionEnvUtil.prepare(parameterTool);
        DataStreamSource<Metrics> data = KafkaConfigUtil.buildSource(env);

        data.addSink(new FlinkKafkaProducer011<Metrics>(
                parameterTool.get("kafka.sink.brokers"),
                parameterTool.get("kafka.sink.topic"),
                new MetricSchema()
                )).name("flink-connectors-kafka")
                .setParallelism(parameterTool.getInt("stream.sink.parallelism"));

        env.execute("flink learning connectors kafka");
    }
}

運(yùn)行結(jié)果

啟動(dòng)程序，查看運(yùn)行結(jié)果婉刀，不段執(zhí)行上面命令吟温，查看是否有新的 topic 出來(lái)：

image

執(zhí)行命令可以查看該 topic 的信息：

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic metric-test

image

分析

上面代碼我們使用 Flink Kafka Producer 只傳了三個(gè)參數(shù)：brokerList、topicId突颊、serializationSchema（序列化）

image

其實(shí)也可以傳入多個(gè)參數(shù)進(jìn)去鲁豪，現(xiàn)在有的參數(shù)用的是默認(rèn)參數(shù)潘悼，因?yàn)檫@個(gè)內(nèi)容比較多，后面可以抽出一篇文章單獨(dú)來(lái)講爬橡。

總結(jié)

本篇文章寫(xiě)了 Flink 讀取其他 Kafka 集群的數(shù)據(jù)治唤，然后寫(xiě)入到本地的 Kafka 上。我在 Flink 這層沒(méi)做什么數(shù)據(jù)轉(zhuǎn)換糙申，只是原樣的將數(shù)據(jù)轉(zhuǎn)發(fā)了下宾添，如果你們有什么其他的需求，是可以在 Flink 這層將數(shù)據(jù)進(jìn)行各種轉(zhuǎn)換操作柜裸，比如這篇文章中的一些轉(zhuǎn)換：《從0到1學(xué)習(xí)Flink》—— Flink Data transformation(轉(zhuǎn)換)缕陕，然后將轉(zhuǎn)換后的數(shù)據(jù)發(fā)到 Kafka 上去。

本文原創(chuàng)地址是: http://www.54tianzhisheng.cn/2019/01/06/Flink-Kafka-sink/ , 未經(jīng)允許禁止轉(zhuǎn)載疙挺。

關(guān)注我

微信公眾號(hào)：zhisheng

另外我自己整理了些 Flink 的學(xué)習(xí)資料扛邑，目前已經(jīng)全部放到微信公眾號(hào)了。你可以加我的微信：zhisheng_tian铐然，然后回復(fù)關(guān)鍵字：Flink 即可無(wú)條件獲取到蔬崩。

image

更多私密資料請(qǐng)加入知識(shí)星球！

image

Github 代碼倉(cāng)庫(kù)

https://github.com/zhisheng17/flink-learning/

以后這個(gè)項(xiàng)目的所有代碼都將放在這個(gè)倉(cāng)庫(kù)里搀暑，包含了自己學(xué)習(xí) flink 的一些 demo 和博客舱殿。

《從0到1學(xué)習(xí)Flink》—— Flink 寫(xiě)入數(shù)據(jù)到 Kafka

《從0到1學(xué)習(xí)Flink》—— Flink 寫(xiě)入數(shù)據(jù)到 Kafka

前言

準(zhǔn)備

添加依賴(lài)

Kafka 安裝

配置文件

程序代碼

運(yùn)行結(jié)果

分析

總結(jié)

關(guān)注我

Github 代碼倉(cāng)庫(kù)

相關(guān)文章

推薦閱讀更多精彩內(nèi)容