4. Flink SQL访问HiveCatalog

一. 实验环境

Flink版本: 1.19.1
Hive版本: 2.1.3
Hadoop版本: 3.2.4

二. 操作步骤

1.上传所需的jar包到Flink lib目录下

[root@hadoop3 ~]# mv flink-sql-connector-hive-3.1.3_2.12-1.19.1.jar /www/flink-1.19.1/lib
[root@hadoop3 ~]# mv hadoop-mapreduce-client-core-3.2.4.jar /www/flink-1.19.1/lib
[root@hadoop3 ~]# mv mysql-connector-java-8.0.30.jar /www/flink-1.19.1/lib

2.修改Hive配置文件hive-site.xml

[root@hadoop3 ~]# vim hive-site.xml

# 配置metastore uris
<property>
    <name>hive.metastore.uris</name>
    <value>thrift://hadoop1:9083</value>
    <description>Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore.</description>
</property>

3.启动Hive外置metastore

[root@hadoop3 ~]# hive --service metastore

3.启动Flink SQL客户端

[root@hadoop3 ~]# yarn-session.sh  -d -nm mysession
[root@hadoop3 ~]# sql-client.sh  -s yarn-session

三. 功能验证

1.Hive创建测试表并插入数据

# 创建测试表
hive> create table t1(id int, name string);
OK
Time taken: 0.106 seconds

# 查看所有表
hive> show tables;
OK
t1
Time taken: 0.056 seconds, Fetched: 1 row(s)

# 插入测试数据
hive> insert into t1 values(1, 'name1'), (2, 'name2');

# 查看测试数据
hive> select * from t1;
OK
1       name1
2       name2
Time taken: 0.229 seconds, Fetched: 2 row(s)

2.Flink创建Catalog并使用

# 创建hive catalog
Flink SQL> CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/www/hive-3.1.3/conf',
	'hadoop-conf-dir'='/www/hadoop-3.2.4/etc/hadoop'
);

3.Flink SQL访问Hive表

Flink SQL> set 'sql-client.execution.result-mode'='tableau';
[INFO] Execute statement succeed.

Flink SQL> show tables;
+------------+
| table name |
+------------+
|         t1 |
+------------+
1 row in set
# 访问hive表数据
Flink SQL> select * from t1;
2025-03-30 10:04:42,897 INFO  org.apache.hadoop.mapred.FileInputFormat                     [] - Total input files to process : 1
2025-03-30 10:04:43,093 WARN  org.apache.flink.yarn.configuration.YarnLogConfigUtil        [] - The configuration directory ('/www/flink-1.19.1/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2025-03-30 10:04:43,125 INFO  org.apache.hadoop.yarn.client.RMProxy                        [] - Connecting to ResourceManager at hadoop1/10.0.49.4:8032
2025-03-30 10:04:43,126 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2025-03-30 10:04:43,133 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface hadoop2:43005 of application 'application_1743044623193_0012'.

+----+-------------+--------------------------------+
| op |          id |                           name |
+----+-------------+--------------------------------+
| +I |           1 |                          name1 |
| +I |           2 |                          name2 |
+----+-------------+--------------------------------+
Received a total of 2 rows (4.61 seconds)