###################################################################################################################

目的

本文是對(duì)參考文獻(xiàn)[1]在高版本上的的復(fù)現(xiàn)

###################################################################################################################

環(huán)境與配置

組件	版本
Flink	1.12
Hive	3.1.2
mysql	8.0.22-0ubuntu0.20.04.2
Zookeeper	3.6.0
Hadoop	3.1.2
Ubuntu	20.04

?

###################################################################################################################

步驟

service firewalld stop(關(guān)閉防火墻)

啓動(dòng)hadoop

離開安全模式

啓動(dòng)zookeeper與kafka集羣

啟動(dòng)flink集群

?該實(shí)驗(yàn)不需要額外的.yaml文件的配置，采用的是DDL方式

操作	命令	備注
查看topic	$KAFKA/bin/kafka-topics.sh --list --zookeeper Desktop:2181	如果想刪除topic,可以是: ? ?
往my_topic 這個(gè) topic發(fā)送 json 消息	$KAFKA/bin/kafka-console-producer.sh --broker-list Desktop:9091 --topic my_topic	這里可能碰到[2]中的報(bào)錯(cuò),注意檢查命令中端口與配置文件server.properties中的listeners的端口嚴(yán)格保持一致 [2]中的報(bào)錯(cuò)還可能是某個(gè)節(jié)點(diǎn)的kafka掛掉導(dǎo)致的. ? 可能碰到[3] 注意關(guān)閉防火墻 ? ?
使用kafka自帶消費(fèi)端測(cè)試下消費(fèi)	$KAFKA/bin/kafka-console-consumer.sh --bootstrap-server Desktop:9091 --from-beginning --topic my_topic	如果kafka自帶消費(fèi)者測(cè)試有問題，那么就不用繼續(xù)往下面做了， 此時(shí)如果使用Flink SQL Client來消費(fèi)也必然會(huì)出現(xiàn)問題
清除topic中所有數(shù)據(jù)[6] (因?yàn)?萬(wàn)一你輸錯(cuò)了呢？對(duì)吧)	$KAFKA/bin/kafka-topics.sh --zookeeper Desktop:2181 --delete --topic my_topic	需要$KAFKA/config/server.properties設(shè)置 delete.topic.enable=true

kafka生產(chǎn)端輸入的數(shù)據(jù)：

{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
{"user_id": "662868", "item_id":"1784", "category_id": "54123654", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
{"user_id": "662854", "item_id":"1456", "category_id": "12345678", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
{"user_id": "662858", "item_id":"1457", "category_id": "12345679", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}
?

###################################################################################################################################################

SQL Client+DDL方式-實(shí)驗(yàn)結(jié)果

?	DDL/SQL	實(shí)驗(yàn)效果
建立表(對(duì)接kafka)	CREATE TABLE user_log1 ( ? ? user_id VARCHAR, ? ? item_id VARCHAR, ? ? category_id VARCHAR, ? ? behavior VARCHAR, ? ? ts VARCHAR ) WITH ( ? ? 'connector.type' = 'kafka', ? ? 'connector.version' = 'universal', ? ? 'connector.topic' = 'my_topic', ? ? 'connector.startup-mode' = 'earliest-offset', ? ? 'connector.properties.group.id' = 'testGroup', ? ? 'connector.properties.zookeeper.connect' = 'Desktop:2181,Laptop:2181,Laptop:2183', ? ? 'connector.properties.bootstrap.servers' = 'Desktop:9091', ? ? 'format.type' = 'json' );
流計(jì)算	select item_id,count(*) from user_log1 group by item_id;

###################################################################################################################################################

Maven工程中嵌入DDL方式

代碼方式參考[2],自己運(yùn)行通過的代碼如下:

https://gitee.com/appleyuchi/Flink_Code/blob/master/FLINK讀寫各種數(shù)據(jù)源/Java/src/main/java/KafkaFlinkDDL.java

?

?

?

?

Reference:

[1]Flink通過SQLClinet創(chuàng)建kafka源表并進(jìn)行實(shí)時(shí)計(jì)算

[2]Flink通過SQLClinet/Java代碼創(chuàng)建kafka源表，指定Offset消費(fèi)，并進(jìn)行實(shí)時(shí)計(jì)算，最后sink到mysql表中

總結(jié)

以上是生活随笔為你收集整理的Flink SQL Client读Kafka+流计算(DDL方式+代碼嵌入DDL/SQL方式)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。