Flink SQL Client方言切换与datagen-＞Hive(DDL形式+streaming形式)

概述

本文是对[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时候卡死-各种情况的解决方案(持续更新中)