Flink 1.12.0集成Hive 3.1.2

前言

Hive 目前已经成为了数据仓库生态系统中的核心组件。它不仅仅是一个用于大数据分析和 ETL 的 SQL 引擎，同样也是一个数据管理平台，它可以用来发现，定义和演变数据。而 Flink 是当前最火的流式计算引擎，它可以在无界和有界数据流上进行状态计算。Flink 从 1.9 版本开始支持集成 Hive，不过 1.9 版本为 Beta 版，不推荐在生产环境中使用。在1.10 版本中，随着对阿里 Blink 的整合完成， Flink 对 Hive 的集成也达到了生产级别的要求。值得注意的是，不同版本的 Flink 对于 Hive 的集成有所差异，本文将以最新的 Flink 1.12 版本、Hive 3.1.2 版本为例，简单阐述 Flink 集成 Hive 的步骤。

集成方式

Flink 与 Hive 的集成包含以下两个层面:

• 首先，Flink 利用了 Hive 的 Metastore 作为持久化的 Catalog，我们可以通过 HiveCatalog 将不同会话中的 Flink 元数据存储到 Hive Metastore 中。例如，我们可以使用 HiveCatalog 将 Kafka 或 Elasticsearch 表的元数据信息存储在 Hive Metastore 中，在后续的 SQL 查询中可以重复使用它们。

• 其次，使用 Flink 来读写 Hive 表，如同使用 SparkSQL 或者 Impala 查询 Hive 中的数据一样。

HiveCatalog 的设计提供了与 Hive 良好的兼容性，用户可以”开箱即用”的访问其已有的 Hive 数仓。不需要修改现有的 Hive Metastore，也不需要更改表的数据位置或分区。

支持的Hive版本

Flink 支持以下 Hive 版本。

请注意，某些功能是否可用取决于您使用的 Hive 版本，这些限制不是由 Flink 所引起的：

• Hive 内置函数在使用 Hive-1.2.0 及更高版本时支持。
• 列约束，也就是 PRIMARY KEY 和 NOT NULL，在使用 Hive-3.1.0 及更高版本时支持。
• 更改表的统计信息，在使用 Hive-1.2.0 及更高版本时支持。
• DATE列统计信息，在使用 Hive-1.2.0 及更高版时支持。
• 使用 Hive-2.0.x 版本时不支持写入 ORC 表。

依赖项

Flink 集成 Hive 需要额外添加一些依赖 jar 包，并将其放在 Flink 安装目录下的 lib 文件夹下，这样才能通过 Table API 或 SQL Client 与 Hive 进行交互。

Flink 官网提供了两种方式添加 Hive 的依赖项。第一种是使用 Flink 提供的 Hive jar，可以根据使用的 Metastore 版本来选择对应的 Hive jar。第二种方式是分别添加每个所需的 jar ，比如说你使用的 Hive 版本与Flink 提供的 Hive jar 兼容的版本不一致。

注意：建议优先使用 Flink 提供的 Hive jar。

使用 Flink 提供的 Hive jar

下面列举了所有可用的 Hive jar，我们可以根据使用的 Hive 版本，下载对应的 jar 即可。比如本文使用的 Hive 版本为 Hive 3.1.2，所以只需要下载 flink-sql-connector-hive-3.1.2.jar 并将其放置在 Flink 安装目录的 lib 文件夹下即可。

用户定义的依赖项

以Hive 3.1.2为例，除了 flink-sql-connector-hive-3.1.2.jar 是我们使用 Flink SQL Cli 时所需要的 jar 外，还需要添加以下 jar：

连接Hive

下面使用 Flink SQL Cli 来连接 Hive。

配置sql-client-defaults.yaml

sql-client-defaults.yaml 文件是 Flink SQL Cli 启动时使用的配置文件，位于 Flink 安装目录的 conf 文件夹下，具体的配置如下，主要是配置catalog：

catalogs:
    - name: myhive
      type: hive
      default-database: default
      hive-conf-dir: /opt/hive/conf/
      hadoop-conf-dir: /opt/hadoop/etc/hadoop/

下表列出了通过 YAML 文件或 DDL 定义 HiveCatalog 时所支持的参数：

操作Hive中的表

首先启动FlinkSQL Cli，命令如下：

sql-client.sh embedded

启动失败，报以下错误：

报错原因是 Flink 集成 Hive 时不支持 embedded metastore，配置 Hive 时需要启动 Hive metastore 服务并在conf/hive-site.xml 配置文件中为 hive.metastore.uris 属性设置正确的值。

启动 Hive metastore 服务

hive --service metastore

在 hive-site.xml 中配置 hive.metastore.uris

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

再次启动 FlinkSQL Cli 成功，接下来我们可以查看注册的catalog

Flink SQL> show catalogs;
default_catalog
myhive

使用注册的myhive catalog

Flink SQL> use catalog myhive;

查询数据库，proghive 是我实践 《Hive编程指南》时创建的 Hive 库

Flink SQL> show databases;
default
proghive

查看所有表

Flink SQL> use proghive;
Flink SQL> show tables;
dividends
employees
stocks

在 Hive 中查询 employees 表：

hive> select * from employees;
OK
John Doe	100000.0	["Mary Smith","Todd Jones"]	{"Federal Taxes":0.2,"State Taxes":0.05,"Insurance":0.1}	{"street":"1 Michigan Ave.","city":"Chicago","state":"IL","zip":60600}
Mary Smith	80000.0	["Bill King"]	{"Federal Taxes":0.2,"State Taxes":0.05,"Insurance":0.1}	{"street":"100 Ontario St.","city":"Chicago","state":"IL","zip":60601}
Todd Jones	70000.0	[]	{"Federal Taxes":0.15,"State Taxes":0.03,"Insurance":0.1}	{"street":"200 Chicago Ave.","city":"Oak Park","state":"IL","zip":60700}
Bill King	60000.0	[]	{"Federal Taxes":0.15,"State Taxes":0.03,"Insurance":0.1}	{"street":"300 Obscure Dr.","city":"Obscuria","state":"IL","zip":60100}
Boss Man	200000.0	["John Doe","Fred Finance"]	{"Federal Taxes":0.3,"State Taxes":0.07,"Insurance":0.05}	{"street":"1 Pretentious Drive.","city":"Chicago","state":"IL","zip":60500}
Fred Finance	150000.0	["Stacy Accountant"]	{"Federal Taxes":0.3,"State Taxes":0.07,"Insurance":0.05}	{"street":"2 Pretentious Drive.","city":"Chicago","state":"IL","zip":60500}
Stacy Accountant	60000.0	[]	{"Federal Taxes":0.15,"State Taxes":0.03,"Insurance":0.1}	{"street":"300 Main St.","city":"Naperville","state":"IL","zip":60563}
Time taken: 0.21 seconds, Fetched: 7 row(s)

现在我们使用 Flink SQL 查询 Hive 中的表。

Flink SQL> select * from employees;

接下来，我们再在 FlinkSQL Cli 中创建一张 Kafka 的数据源表：

CREATE TABLE user_behavior ( 
    `user_id` BIGINT, -- 用户id
    `item_id` BIGINT, -- 商品id
    `cat_id` BIGINT, -- 品类id
    `action` STRING, -- 用户行为
    `province` INT, -- 用户所在的省份
    `ts` BIGINT, -- 用户行为发生的时间戳
    `proctime` AS PROCTIME(), -- 通过计算列产生一个处理时间列
    `eventTime` AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间
     WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND  -- 定义watermark
 ) WITH ( 
    'connector' = 'kafka', -- 使用 kafka connector
    'topic' = 'user_behavior', -- kafka主题
    'scan.startup.mode' = 'earliest-offset', -- 偏移量
    'properties.group.id' = 'group1', -- 消费者组
    'properties.bootstrap.servers' = 'localhost:9092', 
    'format' = 'json', -- 数据源格式为json
    'json.fail-on-missing-field' = 'true',
    'json.ignore-parse-errors' = 'false'
);

查看表结构

Flink SQL> DESCRIBE user_behavior;

我们可以在 Hive 客户端中执行下面命令查看刚刚在 Flink SQL Cli 中创建的表

hive> desc formatted  user_behavior;
OK
# col_name            	data_type           	comment

# Detailed Table Information
Database:           	proghive
OwnerType:          	USER
Owner:              	null
CreateTime:         	Thu Dec 24 15:52:18 CST 2020
LastAccessTime:     	UNKNOWN
Retention:          	0
Location:           	hdfs://localhost:9000/user/hive/warehouse/proghive.db/user_behavior
Table Type:         	MANAGED_TABLE
Table Parameters:
	flink.connector     	kafka
	flink.format        	json
	flink.json.fail-on-missing-field	true
	flink.json.ignore-parse-errors	false
	flink.properties.bootstrap.servers	localhost:9092
	flink.properties.group.id	group1
	flink.scan.startup.mode	earliest-offset
	flink.schema.0.data-type	BIGINT
	flink.schema.0.name 	user_id
	flink.schema.1.data-type	BIGINT
	flink.schema.1.name 	item_id
	flink.schema.2.data-type	BIGINT
	flink.schema.2.name 	cat_id
	flink.schema.3.data-type	VARCHAR(2147483647)
	flink.schema.3.name 	action
	flink.schema.4.data-type	INT
	flink.schema.4.name 	province
	flink.schema.5.data-type	BIGINT
	flink.schema.5.name 	ts
	flink.schema.6.data-type	TIMESTAMP(3) NOT NULL
	flink.schema.6.expr 	PROCTIME()
	flink.schema.6.name 	proctime
	flink.schema.7.data-type	TIMESTAMP(3)
	flink.schema.7.expr 	TO_TIMESTAMP(FROM_UNIXTIME(`ts`, 'yyyy-MM-dd HH:mm:ss'))
	flink.schema.7.name 	eventTime
	flink.schema.watermark.0.rowtime	eventTime
	flink.schema.watermark.0.strategy.data-type	TIMESTAMP(3)
	flink.schema.watermark.0.strategy.expr	`eventTime` - INTERVAL '5' SECOND
	flink.topic         	user_behavior
	is_generic          	true
	transient_lastDdlTime	1608796338

# Storage Information
SerDe Library:      	org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat:        	org.apache.hadoop.mapred.TextInputFormat
OutputFormat:       	org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat
Compressed:         	No
Num Buckets:        	-1
Bucket Columns:     	[]
Sort Columns:       	[]
Storage Desc Params:
	serialization.format	1
Time taken: 0.212 seconds, Fetched: 54 row(s)                

Flink SQL Cli 创建的 user_behavior 表的元数据会持久化到 Hive 的元数据信息库，本文使用的是MySQL。执行下面的命令：

SELECT 
    a.tbl_id, -- 表id
    from_unixtime(create_time) AS create_time, -- 创建时间
    a.db_id, -- 数据库id
    b.name AS db_name, -- 数据库名称
    a.tbl_name -- 表名称
FROM TBLS AS a
LEFT JOIN DBS AS b ON a.db_id =b.db_id
WHERE a.tbl_name = "user_behavior";