流计算 Oceanus 数据仓库 Hive

一、版本说明

二、用作数据目的（Sink）

CREATE TABLE hive_table ( `id` INT, `name` STRING, `dt` STRING, `hr` STRING) PARTITIONED BY (dt, hr)with ( 'connector' = 'hive', — Flink 1.13 请使用 'connector' = 'hive' 'hive-version' = '3.1.1', 'hive-database' = 'testdb', 'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00', 'sink.partition-commit.trigger'='partition-time', 'sink.partition-commit.delay'='1 h', 'sink.partition-commit.policy.kind'='metastore,success-file');

三、作业配置

在 Hive 数据库中创建 Hive 表。

# 在 Hive 的 testdb 数据库创建 hive_table 数据表USE testdb;CREATE TABLE `hive_table` ( `id` int, `name` string)PARTITIONED BY (`dt` string, `hr` string)STORED AS ORC;

对 Hive 表的 HDFS 路径开启写权限。

·方式一：可登录 EMR Hive 集群节点（具体可参见 Hive 基础操作），对目的库 testdb 库的 hive_table 表执行 chmod 操作。

hdfs dfs -chmod 777 /usr/hive/warehouse/testdb.db/hive_table

·方式二：在作业管理 > 作业参数中添加以下高级参数，可以让 Hadoop 用户角色获取 HDFS 路径权限。

containerized.taskmanager.env.HADOOP_USER_NAME: hadoopcontainerized.master.env.HADOOP_USER_NAME: hadoop

注意：说明：

注意：Flink SQL 中使用 Hive 表 testdb.hive_table，这里 CREATE TABLE 的表名对应 Hive 库的表名（Flink 1.13支持通过 hive-table 参数配置覆盖该值），库名通过 hive-database 参数指定。

四、WITH 参数

五、代码示例1

CREATE TABLE datagen_source_table ( id INT, name STRING, log_ts TIMESTAMP(3), WATERMARK FOR log_ts AS log_ts – INTERVAL '5' SECOND) WITH ( 'connector' = 'datagen', 'rows-per-second' = '10');CREATE TABLE hive_table ( `id` INT, `name` STRING, `dt` STRING, `hr` STRING) PARTITIONED BY (dt, hr)with ( 'connector' = 'hive', — Flink 1.13 请使用 'connector' = 'hive' 'hive-version' = '3.1.1', 'hive-database' = 'testdb', 'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00', 'sink.partition-commit.trigger'='partition-time', 'sink.partition-commit.delay'='1 h', 'sink.partition-commit.policy.kind'='metastore,success-file');– streaming sql, insert into hive tableINSERT INTO hive_tableSELECT id, name, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH')FROM datagen_source_table;

六、代码示例2

— hive table 作为source 需要显示指定hive中的database，否则会用默认的数据库default_databaseCREATE database testdb;CREATE TABLE `testdb`.hive_table ( `user_id` INT, `first_name` STRING, `dt` STRING, `hr` STRING ) PARTITIONED BY (dt, hr) with ( 'connector' = 'hive', 'hive-version' = '3.1.1', 'hive-database' = 'testdb', 'partition.time-extractor.timestamp-pattern' = '$dt $hr:00:00', 'streaming-source.enable' = 'true', 'streaming-source.partition.include' = 'all' — all 、 latest, 'streaming-source.monitor-interval' = '1 min', 'streaming-source.partition-order' = 'partition-name' );
CREATE TABLE logger_sink_table ( `user_id` INT, first_name STRING, `dt` STRING, `hr` STRING ) WITH ( 'connector' = 'logger', 'print-identifier' = 'DebugData');INSERT into logger_sink_tableSELECT *from `testdb`.hive_table;

七、获取 Hive 连接配置 jar 包

Flink SQL 任务写 Hive 时需要使用包含 Hive 及 HDFS 配置信息的 jar 包来连接到 Hive 集群。具体获取连接配置 jar 及其使用的步骤如下：

1. ssh 登录到对应 Hive 集群节点。

2. 获取 hive-site.xml 和 hdfs-site.xml，EMR 集群中的配置文件在如下位置。

/usr/local/service/hive/conf/hive-site.xml/usr/local/service/hadoop/etc/hadoop/hdfs-site.xml

3. 修改 hive-site.xml 文件

在hive-site增加如下配置，ip的值取配置文件里 hive.server2.thrift.bind.host 的 value<property> <name>hive.metastore.uris</name> <value>thrift://ip:7004</value></property>

4. 获取 hivemetastore-site.xml 和 hiveserver2-site.xml，点击文件名下载。

5. 对获取到的配置文件 打 jar 包。

jar -cvf hive-xxx.jar hive-site.xml hdfs-site.xml hivemetastore-site.xml hiveserver2-site.xml

6. 校验 jar 的结构（可以通过 vi 命令查看 vi hive-xxx.jar），jar 里面包含如下信息，请确保文件不缺失且结构正确。

META-INF/META-INF/MANIFEST.MFhive-site.xmlhdfs-site.xmlhivemetastore-site.xmlhiveserver2-site.xml

八、在任务中使用配置 jar

引用程序包中选择 Hive 连接配置 jar 包（该 jar 包为在 获取 Hive 连接配置 jar 包 中得到的 hive-xxx.jar，必须在依赖管理上传后才使用）。

九、Kerberos 认证授权

1. 登录集群 Master 节点，获取 krb5.conf、emr.keytab、core-site.xml、hdfs-site.xml、hive-site.xml 文件，路径如下。

/etc/krb5.conf/var/krb5kdc/emr.keytab/usr/local/service/hadoop/etc/hadoop/core-site.xml/usr/local/service/hadoop/etc/hadoop/hdfs-site.xml/usr/local/service/hive/conf/hive-site.xml

2. 修改 hive-site.xml 文件。在 hive-site.xml 中增加如下配置，IP 的值取配置文件中 hive.server2.thrift.bind.host 的 value。

<property> <name>hive.metastore.uris</name> <value>thrift://ip:7004</value></property>

3. 获取 hivemetastore-site.xml 和 hiveserver2-site.xml，点击文件名下载。

4. 对获取的配置文件打 jar 包。

jar cvf hive-xxx.jar krb5.conf emr.keytab core-site.xml hdfs-site.xml hive-site.xml hivemetastore-site.xml hiveserver2-site.xml

5. 校验 jar 的结构（可以通过命令查看 vim hive-xxx.jar），jar 里面包含如下信息，请确保文件不缺失且结构正确。

META-INF/META-INF/MANIFEST.MFemr.keytabkrb5.confhdfs-site.xmlcore-site.xmlhive-site.xmlhivemetastore-site.xmlhiveserver2-site.xml

6. 在 程序包管理 页面上传 jar 包，并在作业参数配置里引用该程序包。

7. 获取 Kerberos principal，用于作业 高级参数 配置。

klist -kt /var/krb5kdc/emr.keytab# 输出如下所示，选取第一个即可：hadoop/172.28.28.51@EMR-OQPO48B9KVNO Timestamp Principal—- ——————- —————————————————— 2 08/09/2021 15:34:40 hadoop/172.28.28.51@EMR-OQPO48B9 2 08/09/2021 15:34:40 HTTP/172.28.28.51@EMR-OQPO48B9 2 08/09/2021 15:34:40 hadoop/VM-28-51-centos@EMR-OQPO48B9 2 08/09/2021 15:34:40 HTTP/VM-28-51-centos@EMR-OQPO48B9

8. 作业 高级参数 配置。

containerized.taskmanager.env.HADOOP_USER_NAME: hadoopcontainerized.master.env.HADOOP_USER_NAME: hadoopsecurity.kerberos.login.principal: hadoop/172.28.28.51@EMR-OQPO48B9security.kerberos.login.keytab: emr.keytabsecurity.kerberos.login.conf: ${krb5.conf.fileName}

注意：历史 Oceanus 集群可能不支持该功能，你可通过 在线客服 联系我们升级集群管控服务，以支持 Kerberos 访问。

十、注意事项

如果 Flink 作业正常运行，日志中没有报错，但是客户端查不到这个 Hive 表，可以使用如下命令对 Hive 表进行修复（需要将 hive_table_xxx 替换为要修复的表名）。

msck repair table hive_table_xxx;