如何在Flink 1.9中使用 Hive？

其中 name 是用户给每个 Catalog 实例指定的名字， Catalog 名字和 DB 名字构成了 FlinkSQL 中元数据的命名空间，因此需要保证每个 Catalog 的名字是唯一的。type 表示 Catalog 的类型，对于 HiveCatalog 而言，type 应该指定为 hive。hive-conf-dir 用于读取 Hive 的配置文件，用户可以将其设定为集群中 Hive 的配置文件目录。hive-version 用于指定所使用的 Hive 版本，可以设定为 2.3.4 或者 1.2.1。

指定了 HiveCatalog 以后，用户就可以启动 sql-client，并通过以下命令验证 HiveCatalog 已经正确加载。

Flink SQL> show catalogs; 
default_catalog 
myhive 
 
Flink SQL> use catalog myhive; 

其中 show catalogs 会列出加载的所有 Catalog 实例。需要注意的是，除了用户在sql-client-defaults.yaml 文件中配置的 Catalog 以外，FlinkSQL 还会自动加载一个 GenericInMemoryCatalog 实例作为内置的 Catalog，该内置 Catalog 默认名字为 default_catalog。

使用 use catalog 可以设定用户 Session 当前的 Catalog。用户在 SQL 语句中访问元数据对象(如 DB、Table 等)时，如果不指定 Catalog 名字，则 FlinkSQL 会在当前 Catalog 中进行查找。

Table API

下面的代码展示了如何通过 TableAPI 来创建 HiveCatalog，并注册到 TableEnvironment。

String name = "myhive"; 
String defaultDatabase = "default"; 
String hiveConfDir = "/path/to/hive_conf_dir"; 
String version = "2.3.4"; 
 
TableEnvironment tableEnv = …; // create TableEnvironment 
HiveCatalog hiveCatalog = new HiveCatalog(name, defaultDatabase, 
hiveConfDir, version); 
tableEnv.registerCatalog(name, hiveCatalog); 
tableEnv.useCatalog(name); 

将 HiveCatalog 注册到 TableEnvironment 以后，就可以在通过 TableEnvironment 提交 SQL 的时候访问 HiveCatalog 中的元数据了。与 SQL Client 类似， TableEnvironment 也提供了 useCatalog 接口让用户设定当前 Catalog。

读写 Hive 表

设置好 HiveCatalog 以后就可以通过 SQL Client 或者 Table API 来读写 Hive 中的表了。

SQL Client

假设 Hive 中已经有一张名为 src 的表，我们可以用以下的 SQL 语句来读写这张表。

Flink SQL> describe src; 
root 
 |-- key: STRING 
 |-- value: STRING 
 
 
Flink SQL> select * from src; 
 
                      key                     value 
                       100                   val_100 
                       298                   val_298 
                         9                     val_9 
                       341                   val_341 
                       498                   val_498 
                       146                   val_146 
                       458                   val_458 
                       362                   val_362 
                       186                   val_186 
                       ……                   …… 
 
Flink SQL> insert into src values ('newKey','newVal'); 

Table API

（编辑：威海站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!