flink连接hive Demon_flink1.17 hive

HiveCatalog

HiveCatalog 有两个用途：作为原生 Flink 元数据的持久化存储，以及作为读写现有 Hive 元数据的接口。 Flink 的 Hive 文档 提供了有关设置 HiveCatalog 以及访问现有 Hive 元数据的详细信息。

导入需要的依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
<!-- Hive Dependency -->
<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-exec</artifactId>
    <version>3.1.2</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
11

启动hive元数据

nohup hive --service metastore >/dev/null 2>&1 &
1

连接 Hive

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

String name            = "myhive";  // Catalog名字
String defaultDatabase = "flink_test"; //默认数据库
String hiveConfDir     = "c:/conf"; // hive配置文件的目录.需要把hive-site.xml添加到该目录
// 1.创建HiveCatalog
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
// 2.注册HiveCatalog
tEnv.registerCatalog(name, hive);
// 3.把HiveCatalog: myhive作为当前session的catalog
tEnv.useCatalog(name);
tEnv.useDatabase("flink_test");
tEnv.sqlQuery("select  from stu").execute().print();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15