概述

本文是對[1]的完整復現,補充了[1]中缺失的大量細節.

?

切換方言

切換目標	Flink SQL Client命令
切換為hive	SET table.sql-dialect=hive;
切換為flink	SET table.sql-dialect=default;

?

Flink SQL Client中方言切換為hive的時候,指的是對hive進行實際操作,此時不再單純建立sink而已.

整個實驗流程

數據類型:

streaming

環境與準備工作

組件	版本
Hadoop	3.1.2
Zookeeper	3.6.0
Flink	1.12
Hive	2.3.6
flink-sql-connector-hive	2.3.6

flink-sql-connector-hive最高版本是2.3.6,所以盡量還是不要弄個2.3.7的hive版本了

嚴格檢查版本能省去有一大堆網上找都找不到的運維問題(詳見附錄)

本實驗需要yaml文件配合

https://gitee.com/appleyuchi/Flink_Code/blob/master/FLINK讀寫各種數據源/flink-hive-streaming.yaml

把上面的配置文件放到$FLINK_HOME/conf下面

需要啟動hadoop,zookeeper,flink四種集群

需要啟動hive 的元服務.

hive的客戶端beeline需要啟動hiveserver2才能正常使用

?

版本檢查

檢查下面5個地方

$FLINK_HOME/conf/*.yaml(至少兩處:version以及 conf的路徑)

$HIVE_HOME(檢查hive版本)

$FLINK_HOME/lib下面的flink-sql-connector-hive-2.3.6_2.12-1.12.0.jar(注意這個jar里面的版本號)

Flink版本盡量1.11以上,因為最近變動有點大.

$FLINK_HOME/lib下面的hive-common-2.3.6.jar/hive-exec-2.3.6.jar版本盡量要與使用的hive版本一致
?

?

完整實驗步驟

步驟	效果
啟動(具體見下方sql文件)	-
加載catalog(具體見下方sql文件)	-
建立source(具體見下方sql文件)	-
切換方言為hive 建立sink(具體見下方sql文件)	-
通過insert語句提交任務
在hive中查看Flink SQL寫入的流數據

完整SQL操作如下:

https://gitee.com/appleyuchi/Flink_Code/blob/master/FLINK讀寫各種數據源/flink-hive-streaming.sql

?

注意事項:

如果忘記使用use catalog myhive;

最后會導致雖然成功建立了一個hive的sink table,但是會報錯[5]

所以千萬不要忘記.

?

?

附錄

我碰到一些運維的問題[4][5][6][7]

關于flink讀取hive卡住的問題[8]

以及hive正常操作時卡住的問題參考[9]

?

?

Reference:

[1]Flink Sql on Zeppelin（6）——Hive Streaming

[2]Flink集成Hive之快速入門--以Flink1.12為例

[3]Hive Read & Write

[4]Querying an unbounded table ‘myhive.db1.datagen‘ in batch mode is not allowed

[5]Table options do not contain an option key ‘connector‘ for discovering a connector

[6]org.apache.flink.table.catalog.exceptions.CatalogException: Failed to create Hive Metastore client

[7]org.apache.hadoop.hive.metastore.api.SerDeInfo； local class incompatible

[8]flink sql client讀取hive時卡住

[9]Hive drop table時候卡死-各種情況的解決方案(持續更新中)

總結

以上是生活随笔為你收集整理的Flink SQL Client方言切换与datagen-＞Hive(DDL形式+streaming形式)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。