SQLClint整合Hive

启动服务（MySQL、HDFS和Hive MetaStore），配置Flink SQL Client 集成Hive，步骤如下：

• 启动服务

xxxxxxxxxx
# 1. 确定MySQL数据库启动(自动启动忽略此步)
[root@node1 ~]# systemctl start mysqld
# 2. 启动HDFS服务（集群）
[root@node1 ~]# starthdfs.sh    
# 3. 启动HiveMetaStore服务
[root@node3 ~]# nohup hive --service metastore &

• 将软件\集成hive依赖jar包下的jar上传到node1、node2、node3的/opt/flink-salone/lib/目录

• node1上重新启动Flink Cluster集群

xxxxxxxxxx
/opt/flink-salone/bin/stop-cluster.sh 
export HADOOP_CLASSPATH=`hadoop classpath`
/opt/flink-salone/bin/start-cluster.sh

• node2上备份hive-site.xml

xxxxxxxxxx
[root@node2 ~]# cd /opt/hive-3.1.2/conf/
[root@node2 conf]# cp hive-site.xml hive-site.xml_backup_flink_and_hive

• 将node4上的hive-site.xml文件远程拷贝到node2上

xxxxxxxxxx
[root@node4 ~]# cd /opt/hive-3.1.2/conf/
[root@node4 conf]# scp hive-site.xml node2:`pwd`

• node2启动SQL Client命令行，并设置配置参数

xxxxxxxxxx
/opt/flink-salone/bin/sql-client.sh embedded
# 设置配置参数
SET sql-client.execution.mode=batch;
SET sql-client.execution.result-mode=tableau;
SET parallism.default=1;

• 创建Catalog:

xxxxxxxxxx
CREATE CATALOG hive_catalog WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/opt/hive-3.1.2/conf/',
    'hive-version' = '3.1.2',
    'hadoop-conf-dir' = '/opt/hadoop-3.1.3/etc/hadoop/'
);

• 查看所有的catalog：

xxxxxxxxxx
Flink SQL> show catalogs;
+-----------------+
|    catalog name |
+-----------------+
| default_catalog |
|    hive_catalog |
+-----------------+
2 rows in set

• 删除catalog

xxxxxxxxxx
CREATE CATALOG testcatalog WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/opt/hive-3.1.2/conf/',
    'hive-version' = '3.1.2',
    'hadoop-conf-dir' = '/opt/hadoop-3.1.3/etc/hadoop/'
);
DROP CATALOG testcatalog

• 使用Catalog:

xxxxxxxxxx
USE CATALOG hive_catalog;

• 创建数据库实例：

xxxxxxxxxx
Flink SQL> CREATE DATABASE IF NOT EXISTS hive_catalog.flink_sql_db
COMMENT 'flink sql database';
[INFO] Execute statement succeed.
Flink SQL> use flink_sql_db;
[INFO] Execute statement succeed.
Flink SQL> show tables;
Empty set