如何進(jìn)行數(shù)據(jù)sink到kafka的操作
這篇文章給大家介紹如何進(jìn)行數(shù)據(jù)sink到kafka的操作，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。
創(chuàng)新互聯(lián)專注于臨滄企業(yè)網(wǎng)站建設(shè),成都響應(yīng)式網(wǎng)站建設(shè),商城網(wǎng)站建設(shè)。臨滄網(wǎng)站建設(shè)公司,為臨滄等地區(qū)提供建站服務(wù)。全流程按需開(kāi)發(fā)，專業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，創(chuàng)新互聯(lián)專業(yè)和態(tài)度為您提供的服務(wù)
下面來(lái)體驗(yàn)將數(shù)據(jù)sink到kafka的操作。
版本和環(huán)境準(zhǔn)備
本次實(shí)戰(zhàn)的環(huán)境和版本如下：
JDK：1.8.0_211
Flink：1.9.2
Maven：3.6.0
操作系統(tǒng)：macOS Catalina 10.15.3 （MacBook Pro 13-inch, 2018）
IDEA：2018.3.5 (Ultimate Edition)
Kafka：2.4.0
Zookeeper：3.5.5
<font color="red">請(qǐng)確保上述環(huán)境和服務(wù)已經(jīng)就緒；</font>
源碼下載
如果您不想寫代碼，整個(gè)系列的源碼可在GitHub下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：
| 名稱 | 鏈接 | 備注 |
|---|---|---|
| 項(xiàng)目主頁(yè) | https://github.com/zq2599/blog_demos | 該項(xiàng)目在GitHub上的主頁(yè) |
| git倉(cāng)庫(kù)地址(https) | https://github.com/zq2599/blog_demos.git | 該項(xiàng)目源碼的倉(cāng)庫(kù)地址，https協(xié)議 |
| git倉(cāng)庫(kù)地址(ssh) | git@github.com:zq2599/blog_demos.git | 該項(xiàng)目源碼的倉(cāng)庫(kù)地址，ssh協(xié)議 |
這個(gè)git項(xiàng)目中有多個(gè)文件夾，本章的應(yīng)用在<font color="blue">flinksinkdemo</font>文件夾下，如下圖紅框所示： 
準(zhǔn)備完畢，開(kāi)始開(kāi)發(fā)；
準(zhǔn)備工作
正式編碼前，先去官網(wǎng)查看相關(guān)資料了解基本情況：
地址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html
我這里用的kafka是2.4.0版本，在官方文檔查找對(duì)應(yīng)的庫(kù)和類，如下圖紅框所示：
kafka準(zhǔn)備
創(chuàng)建名為test006的topic，有四個(gè)分區(qū)，參考命令：
./kafka-topics.sh \ --create \ --bootstrap-server 127.0.0.1:9092 \ --replication-factor 1 \ --partitions 4 \ --topic test006
在控制臺(tái)消費(fèi)test006的消息，參考命令：
./kafka-console-consumer.sh \ --bootstrap-server 127.0.0.1:9092 \ --topic test006
此時(shí)如果該topic有消息進(jìn)來(lái)，就會(huì)在控制臺(tái)輸出；
接下來(lái)開(kāi)始編碼；
創(chuàng)建工程
用maven命令創(chuàng)建flink工程：
mvn \ archetype:generate \ -DarchetypeGroupId=org.apache.flink \ -DarchetypeArtifactId=flink-quickstart-java \ -DarchetypeVersion=1.9.2
根據(jù)提示，groupid輸入<font color="blue">com.bolingcavalry</font>，artifactid輸入<font color="blue">flinksinkdemo</font>，即可創(chuàng)建一個(gè)maven工程；
在pom.xml中增加kafka依賴庫(kù)：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>1.9.0</version> </dependency>
工程創(chuàng)建完成，開(kāi)始編寫flink任務(wù)的代碼；
發(fā)送字符串消息的sink
先嘗試發(fā)送字符串類型的消息：
創(chuàng)建KafkaSerializationSchema接口的實(shí)現(xiàn)類，后面這個(gè)類要作為創(chuàng)建sink對(duì)象的參數(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));
}
}創(chuàng)建任務(wù)類KafkaStrSink，請(qǐng)注意FlinkKafkaProducer對(duì)象的參數(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)建一個(gè)List，里面有兩個(gè)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)計(jì)每個(gè)單詞的數(shù)量
env.fromCollection(list)
.addSink(producer)
.setParallelism(4);
env.execute("sink demo : kafka str");
}
}使用mvn命令編譯構(gòu)建，在target目錄得到文件<font color="blue">flinksinkdemo-1.0-SNAPSHOT.jar</font>；
在flink的web頁(yè)面提交flinksinkdemo-1.0-SNAPSHOT.jar，并制定執(zhí)行類，如下圖：
提交成功后，如果flink有四個(gè)可用slot，任務(wù)會(huì)立即執(zhí)行，會(huì)在消費(fèi)kafak消息的終端收到消息，如下圖：
任務(wù)執(zhí)行情況如下圖：
發(fā)送對(duì)象消息的sink
再來(lái)嘗試如何發(fā)送對(duì)象類型的消息，這里的對(duì)象選擇常用的Tuple2對(duì)象：
創(chuàng)建KafkaSerializationSchema接口的實(shí)現(xiàn)類，該類后面要用作sink對(duì)象的入?yún)?，?qǐng)注意代碼中捕獲異常的那段注釋：<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)境這是個(gè)非常危險(xiǎn)的操作，
// 過(guò)多的錯(cuò)誤打印會(huì)嚴(yán)重影響系統(tǒng)性能，請(qǐng)根據(jù)生產(chǎn)環(huán)境情況做調(diào)整
e.printStackTrace();
}
return new ProducerRecord<byte[], byte[]>(topic, b);
}
}創(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)建一個(gè)List，里面有兩個(gè)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)計(jì)每個(gè)單詞的數(shù)量
env.fromCollection(list)
.keyBy(0)
.sum(1)
.addSink(producer)
.setParallelism(4);
env.execute("sink demo : kafka obj");
}
}像前一個(gè)任務(wù)那樣編譯構(gòu)建，把jar提交到flink，并指定執(zhí)行類是<font color="blue">com.bolingcavalry.addsink.KafkaObjSink</font>；
消費(fèi)kafka消息的控制臺(tái)輸出如下：
在web頁(yè)面可見(jiàn)執(zhí)行情況如下：
至此，flink將計(jì)算結(jié)果作為kafka消息發(fā)送出去的實(shí)戰(zhàn)就完成了，希望能給您提供參考，接下來(lái)的章節(jié)，我們會(huì)繼續(xù)體驗(yàn)官方提供的sink能力；
關(guān)于如何進(jìn)行數(shù)據(jù)sink到kafka的操作就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。
當(dāng)前標(biāo)題：如何進(jìn)行數(shù)據(jù)sink到kafka的操作
本文來(lái)源：http://bm7419.com/article16/pciigg.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供響應(yīng)式網(wǎng)站、Google、用戶體驗(yàn)、、商城網(wǎng)站、自適應(yīng)網(wǎng)站
聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)
- 原創(chuàng)和偽原創(chuàng)如何影響網(wǎng)站排名 2021-12-13
- 【精華】有哪些值得借鑒恢復(fù)網(wǎng)站排名方法 2022-06-24
- 如何快速提升成都網(wǎng)絡(luò)優(yōu)化網(wǎng)站排名？ 2022-09-27
- 做好網(wǎng)站排名優(yōu)化工作，新手優(yōu)化人員要知道的小細(xì)節(jié)有哪些？ 2022-05-01
- 做SEC如何提高網(wǎng)站排名 SEC有哪些手段呢 2015-06-16
- 合理的蜘蛛程序發(fā)開(kāi)有利于網(wǎng)站排名優(yōu)化 2023-04-04
- 網(wǎng)站排名優(yōu)化技巧透析 2017-03-03
- 如何讓網(wǎng)站排名更靠前 2016-11-02
- 掌握這八大門戶網(wǎng)站優(yōu)化技巧，網(wǎng)站排名不上去都難！ 2014-10-30
- 百度收錄越多，網(wǎng)站排名就越高嗎？ 2018-03-06
- 網(wǎng)站排名不穩(wěn)定是什么原因造成的 2013-12-25
- 網(wǎng)站排名高級(jí)關(guān)鍵詞工具研究 2014-12-26