流计算 Oceanus 使用 Hive Catalog

一、版本说明

二、前提条件

已在 Hive Metastore 侧开启了 Hive Metastore 服务。
相关命令如下：

·hive –service metastore：开启 Hive Metastore 服务。

·ps -ef|grep metastore：查询 Hive Metastore 服务是否已开启。

三、创建 Hive Catalog

切换到 _dc，单击新建 Hive Catalog。

上传4个配置文件 hive-site.xml ，hdfs-site.xml ，hivemetastore-site.xml ，hiveserver2-site.xml 配置文件下载，其中 hive-site.xml 中需要添加 urls 的地址。

四、创建数据库

在 SQL 作业中可以创建数据库。数据库的写法采用两段式：catalog_name.database_name。

CREATE DATABASE IF NOT EXISTS `hiveCatalogName`.`databaseName`;

五、创建数据表

在 SQL 作业中可以创建数据表。数据表的写法采用三段式：catalog_name.database_name.table_name。

CREATE TABLE IF NOT EXISTS `hiveCatalogName`.`databaseName`.`tableName` ( user_id INT, item_id INT, category_id INT, — ts AS localtimestamp, — WATERMARK FOR ts AS ts, behavior VARCHAR) WITH ( 'connector' = 'datagen', 'rows-per-second' = '1', — 每秒产生的数据条数 'fields.user_id.kind' = 'sequence', — 有界序列（结束后自动停止输出） 'fields.user_id.start' = '1', — 序列的起始值 'fields.user_id.end' = '10000', — 序列的终止值 'fields.item_id.kind' = 'random', — 无界的随机数 'fields.item_id.min' = '1', — 随机数的最小值 'fields.item_id.max' = '1000', — 随机数的最大值 'fields.category_id.kind' = 'random', — 无界的随机数 'fields.category_id.min' = '1', — 随机数的最小值 'fields.category_id.max' = '1000', — 随机数的最大值 'fields.behavior.length' = '5' — 随机字符串的长度);

六、SQL 作业中引用 Hive Catalog 中的表

在 SQL 作业中，将光标移动到要插入的元表的位置，在左侧导航栏中找到要引用的表，点击菜单中的【引用】。

INSERT INTO `hiveCatalogName`.`databaseName`.`sink_tableName` SELECT *FROM `hiveCatalogName`.`databaseName`.`source_tableName`;

Table hints（可选）

通过 Hints 动态覆盖表的配置项，无需修改表定义即可临时调整参数。

示例：对 Source 表指定开启流式读取。

— Flink 1.13需要开启动态表选项（其他Flink版本不需要）SET 'table.dynamic-table-options.enabled' = 'true';INSERT INTO `hiveCatalogName`.`databaseName`.`sink_tableName`SELECT *FROM `hiveCatalogName`.`databaseName`.`source_tableName` /*+ OPTIONS( 'streaming-source.enable' = 'true', 'streaming-source.monitor-interval' = '1min', 'streaming-source.consume-order' = 'create-time' ) */;

注意：说明：

七、删除 Hive Metastore

在左侧导航栏，单击 Hive Catalog 对应的删除按钮。


八、赋权

作业执行过程中，需要有写入 HDFS 文件的权限，目前 Oceanus 使用 Hive Catalog 元数据需要对 Flink 用户进行授权。相关操作如下：

·在 Hive 所在的所有 master 节点执行。

useradd flinkgroupadd supergroupusermod -a -G supergroup flinkhdfs dfsadmin -refreshUserToGroupsMappings

·建议在 Hive 中开启权限，在 hive-site.xml 文件中添加如下的配置项。

<property><name>hive.metastore.authorization.storage.checks</name><value>true</value><description>Should the metastore do authorization checks againstthe underlying storage for operations like drop-partition (disallowthe drop-partition if the user in question doesn't have permissionsto delete the corresponding directory on the storage).</description><property>