實驗?zāi)康?/h3>

Kafka的數(shù)據(jù)能讓Flink SQL Client讀取到

本文是對[1]的詳細(xì)記載

具體操作步驟

①啓動hadoop集羣,離開安全模式

②各個節(jié)點都關(guān)閉防火墻:

service firewalld status(查看防火墻狀態(tài))

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

各個節(jié)點分別啟動zookeeper

③啟動kafka集群

startkafka

startkafka2

startkafka3

④

flink-connector-kafka_2.12-1.12.0.jar

flink-json-1.12.0.jar

flink-jdbc_2.12-1.10.2.jar

放入$FLINK_HOME/lib中

啟動flink集群

?

?

⑤

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

?

下面的需要手動輸入(發(fā)送json消息,注意下面的信息千萬不要一次性復(fù)制全部內(nèi)容,必須一條一條手動拷貝)

{"order_id": "1","shop_id": "AF18","member_id": "3410211","trade_amt": "100.00","pay_time": "1556420980000"}
{"order_id": "2","shop_id": "AF20","member_id": "3410213","trade_amt": "130.00","pay_time": "1556421040000"}
{"order_id": "3","shop_id": "AF18","member_id": "3410212","trade_amt": "120.00","pay_time": "1556421100000"}
{"order_id": "4","shop_id": "AF19","member_id": "3410212","trade_amt": "100.00","pay_time": "1556421120000"}
{"order_id": "5","shop_id": "AF18","member_id": "3410211","trade_amt": "150.00","pay_time": "1556421480000"}
{"order_id": "6","shop_id": "AF18","member_id": "3410211","trade_amt": "110.00","pay_time": "1556421510000"}
{"order_id": "7","shop_id": "AF19","member_id": "3410213","trade_amt": "110.00","pay_time": "1556421570000"}
{"order_id": "8","shop_id": "AF20","member_id": "3410211","trade_amt": "100.00","pay_time": "1556421630000"}
{"order_id": "9","shop_id": "AF17","member_id": "3410212","trade_amt": "110.00","pay_time": "1556421655000"}
?

⑥

$FLINK_HOME/bin/sql-client.sh embedded -d $FLINK_HOME/conf/sql.my.yaml -l sql-libs/

注意這個實驗涉及到兩個.yaml文件，修改如下:

?

?

$FLINK_HOME/bin/sql-client.sh embedded -d $FLINK_HOME/conf/sql.my.yaml -l /home/appleyuchi/bigdata/flink-1.12/lib

⑦FLINK SQL執(zhí)行

具體操作	具體FLINK SQL
顯示orders內(nèi)容	select * from orders;
1分鐘固定窗口計算	SELECT ? shop_id ? , TUMBLE_START(payment_time, INTERVAL '1' MINUTE) AS tumble_start ? , TUMBLE_END(payment_time, INTERVAL '1' MINUTE) ? AS tumble_end ? , sum(trade_amt) ? ? ? ? ? ? ? ? ? ? ? ? ? ? AS amt FROM orders GROUP BY shop_id, TUMBLE(payment_time, INTERVAL '1' MINUTE);

?

?

--------------------------------------------------------------------------------------------------------實驗效果截圖--------------------------------------------------------------------------------------------------------------------------------------------------------------------

解決方案：

$FLINK_HOME/conf/flink-conf.yaml中修改為:

classloader.resolve-order: parent-first

盡量集群中的每個節(jié)點都要修改

繼續(xù)往下，碰到問題:

問了花名雪盡

根據(jù)提示：

放棄采用.yaml的方式，改用DDL方式(下面的直接拷貝到Flink SQL Client中然后按下回車即可):

CREATE TABLE orders (order_id BIGINT, member_id BIGINT, trade_amt DOUBLE, pay_time BIGINT,ts AS TO_TIMESTAMP(FROM_UNIXTIME(pay_time/ 1000, 'yyyy-MM-dd HH:mm:ss')), -- 定義事件時間WATERMARK FOR ts AS ts - INTERVAL '5' SECOND -- 在ts上定義5 秒延遲的 watermark ) WITH ('connector.type' = 'kafka','connector.version' = 'universal','connector.topic' = 'order_sql','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' );

操作如下:

實驗效果如下:

----------------------------------------------------------------------------附錄（其他可能用到的常規(guī)操作）-----------------------------------------------------------------------------------------------------------------------------------------

?

刪除topic:

$KAFKA/bin/kafka-topics.sh --delete --zookeeper Desktop:2181 --topic order_sql

或者:

deleteall /brokers/topics/order_sql
deleteall /config/topics/order_sql
deleteall /admin/delete_topics/order_sql

?

?

?

根據(jù)官方文檔[4]用的是0.11的kafka，可能是版本存在不兼容的問題

Reference:

[1]Flink SQL-Client 的使用

[2]Kafka連接服務(wù)器出現(xiàn):Connection to node 1 (localhost/127.0.0.1:9092) could not be established.

[3]kafka出現(xiàn)Unable to read additional data from server sessionid 0x0, likely server has closed socket

[4]SQL Client

[5]Flink 1.9 實戰(zhàn)：使用 SQL 讀取 Kafka 并寫入 MySQL

[6]Is there a way to delete all the data from a topic or delete the topic before every run?

?

?

?

?

?

總結(jié)

以上是生活随笔為你收集整理的flink sql client讀取kafka數據的timestamp(DDL方式)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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