flink-sql读写hive-1.16

1. 版本说明

本文档内容基于flink-1.16.x，其他版本的整理，请查看本人博客的 flink 专栏其他文章。

1.1. 概述

Apache Hive 已经成为了数据仓库生态系统中的核心。它不仅仅是一个用于大数据分析和ETL场景的SQL引擎，同样也是一个数据管理平台，可用于发现，定义，和演化数据。

Flink 与 Hive 的集成包含两个层面。

一是利用了 Hive 的 MetaStore 作为持久化的 Catalog，用户可通过HiveCatalog将不同会话中的 Flink 元数据存储到 Hive Metastore 中。
例如，用户可以使用HiveCatalog将Kafka表或 Elasticsearch 表存储在 Hive Metastore 中，并后续在 SQL 查询中重新使用它们。

二是利用 Flink 来读写 Hive 表。

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

1.2. 支持的Hive版本

Flink 支持以下的 Hive 版本。

• 1.0
  • 1.0.0
  • 1.0.1
• 1.1
  • 1.1.0
  • 1.1.1
• 1.2
  • 1.2.0
  • 1.2.1
  • 1.2.2
• 2.0
  • 2.0.0
  • 2.0.1
• 2.1
  • 2.1.0
  • 2.1.1
• 2.2
  • 2.2.0
• 2.3
  • 2.3.0
  • 2.3.1
  • 2.3.2
  • 2.3.3
  • 2.3.4
  • 2.3.5
  • 2.3.6
  • 2.3.7
  • 2.3.8
  • 2.3.9
• 3.1
  • 3.1.0
  • 3.1.1
  • 3.1.2
  • 3.1.3

注意，某些功能是否可用取决于使用的 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 表。

1.2.1. 依赖项

要与 Hive 集成，需要在 Flink 下的 /lib 目录中添加一些额外的依赖包，以便通过 Table API 或 SQL Client 与 Hive 进行交互。
也可以将这些依赖项放在专用文件夹中，并分别使用 Table API 程序或 SQL Client 的-C或-l选项将它们添加到 classpath 中。

Apache Hive 是基于 Hadoop 之上构建的，因此需要通过设置HADOOP_CLASSPATH环境变量来配置hadoop的依赖项：

export HADOOP_CLASSPATH=`hadoop classpath`

有两种添加 Hive 依赖项的方法。第一种是使用 Flink 提供的 Hive Jar 包。可以根据使用的 Metastore 版本来选择对应的 Hive jar。第二个方式是分别添加每个所需的 jar 包。
如果使用的 Hive 版本尚未在此处列出，则第二种方法会更适合。

注意：建议优先使用 Flink 提供的 Hive jar 包。仅在 Flink 提供的 Hive jar 不满足需求时，再考虑使用分开添加 jar 包的方式。

1.2.1.1. 使用 Flink 提供的 Hive jar

下表列出了所有可用的 Hive jar。您可以选择一个并放在 Flink 发行版的 /lib 目录中。

1.2.1.2. 使用用户定义的依赖项

可以在这儿找到不同Hive主版本所需要的依赖项。

1.2.1.3. 移除计划器jar

Move the jar flink-table-planner_2.12-1.16.0.jar located in FLINK_HOME/opt to FLINK_HOME/lib and move out the jar flink-table-planner-loader-1.16.0.jar from FLINK_HOME/lib. Please refer to FLINK-25128 for more details. You can use the following commands to complete the work:

将 FLINK_HOME/lib 目录下的 flink-table-planner-loader-1.16.0.jar 移动到 FLINK_HOME/opt 目录，然后将 FLINK_HOME/opt 目录下的 flink-table-planner_2.12-1.16.0.jar 移动到 FLINK_HOME/lib 目录下。可以参考 FLINK-25128 来了解更多信息。可以使用下面的命令来完成这个操作。

mv $FLINK_HOME/opt/flink-table-planner_2.12-1.16.0.jar $FLINK_HOME/lib/flink-table-planner_2.12-1.16.0.jar
mv $FLINK_HOME/lib/flink-table-planner-loader-1.16.0.jar $FLINK_HOME/opt/flink-table-planner-loader-1.16.0.jar

注意：只有当使用 Hive dialect 或 HiveServer2 时需要这些额外的计划器 jar 包，但是为了整合 hive ，建议执行上面的操作。

1.2.2. maven 程序依赖

如果你在构建自己的程序，你需要将下面的依赖添加到你的 pom 文件中，不过并不需要将这些依赖打包到最终的 jar 文件中，可以在运行时将上面的依赖添加 flink 环境中。

<!-- Flink Dependency -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hive_2.12</artifactId>
  <version>1.16.0</version>
  <scope>provided</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-java-bridge_2.12</artifactId>
  <version>1.16.0</version>
  <scope>provided</scope>
</dependency>

<!-- Hive Dependency -->
<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-exec</artifactId>
    <version>${hive.version}</version>
    <scope>provided</scope>
</dependency>

1.3. 连接到hive

可以在表环境或 YAML 配置中使用 catalog interface 和 HiveCatalog 去连接一个已经存在的 hive 实例。

下面是连接 hive 的一个案例：

java

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);

String name            = "myhive";
String defaultDatabase = "mydatabase";
String hiveConfDir     = "/opt/hive-conf";

HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog("myhive", hive);

// 设置 HiveCatalog 为会话的当前 catalog
tableEnv.useCatalog("myhive");

scala

val settings = EnvironmentSettings.newInstance().useBlinkPlanner().build()
val tableEnv = TableEnvironment.create(settings)

val name            = "myhive"
val defaultDatabase = "mydatabase"
val hiveConfDir     = "/opt/hive-conf"

val hive = new HiveCatalog(name, defaultDatabase, hiveConfDir)
tableEnv.registerCatalog("myhive", hive)

// 设置 HiveCatalog 为会话的当前 catalog
tableEnv.useCatalog("myhive")

python

from pyflink.table import *
from pyflink.table.catalog import HiveCatalog

settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()
t_env = TableEnvironment.create(settings)

catalog_name = "myhive"
default_database = "mydatabase"
hive_conf_dir = "/opt/hive-conf"

hive_catalog = HiveCatalog(catalog_name, default_database, hive_conf_dir)
t_env.register_catalog("myhive", hive_catalog)

# 设置 HiveCatalog 为会话的当前 catalog
tableEnv.use_catalog("myhive")

yaml

execution:
    planner: blink
    ...
    current-catalog: myhive  # 设置 HiveCatalog 为会话的当前 catalog
    current-database: mydatabase
    
catalogs:
   - name: myhive
     type: hive
     hive-conf-dir: /opt/hive-conf

sql

CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'mydatabase',
    'hive-conf-dir' = '/opt/hive-conf'
);
-- 设置 HiveCatalog 为会话的当前 catalog
USE CATALOG myhive;

下面是使用 YAML 文件或 DDL 语句创建 HiveCatalog 实例时支持的一些参数。

1.4. DDL

建议在 flink 中使用 hive 方言执行 DDL 来创建 hive 表、视图、分区、函数。

1.5. DML

flink 支持使用 DML 语句写 hive 表。请参考 Reading & Writing Hive Tables。

2. Hive Catalog

最近几年，在hadoop生态系统中，Hive Metastore已经成为事实上的元数据中心。很多公司都有一个Hive Metastore服务示例作为他们的产品来管理所有的元数据，不管这些元数据是hive，还是非hive的。

对于同时有hive和flink的部署，HiveCatalog 可以让集群使用 Hive Metastore 来管理flink的元数据。

对于只有 flink 的部署，HiveCatalog 是 flink 唯一开箱即用的持久化 catalog。如果不使用持久化catalog，用户使用 Flink SQL CREATE DDL 创建诸如kafka表时，必须在每个会话中重复创建元数据对象，这会浪费很多时间。
HiveCatalog可以通过授权用户值创建一次表和其他元数据对象来填补这个空隙，并且稍后就可以在所有会话中使用和管理他们。

2.1. 设置HiveCatalog

2.1.1. 依赖

在 flink 中设置 HiveCatalog 要求在所有的 Flink-Hive 集成环境中有相同的依赖。

2.1.2. 配置

在 flink 中设置 HiveCatalog 要求在所有的 Flink-Hive 集成环境中有相同的配置。

2.2. 使用HiveCatalog

一旦配置成功，HiveCatalog就可以开箱即用。用户可以使用DDL来创建flink的元数据对象，并且马上就可以看到他们。

HiveCatalog可以处理两种类型的表：与hive兼容的表和通用表。

• hive兼容表是通过hive存储的表，他们的元数据和实际的数据都在分层存储中。因此，通过flink创建的与hive兼容的表，可以通过hive查询。
• 通用表是特定于flink处理的。当使用HiveCatalog创建通用表时，我们必须使用HMS（hive matestore）来持久化元数据。虽然这些表在hive可见，但这并不意味着hive可以理解这些元数据。
  因此在hive中使用这些表，会导致未定义的行为。一般只通过hive来查看这些表的schema元数据。

flink使用is_generic来描述一张表是hive兼容表还是通用表。当使用HiveCatalog创建表时，默认为通用表。如果想要创建一个hive兼容表，则需要在表的属性中显示设置is_generic为false。

就像上面描述的一样，通用表不能在hive中使用。在 hive 客户端中，你可以使用 DESCRIBE FORMATTED 语句来查看表元数据中的 is_generic 属性来判断该表是否为通用表。通用表的属性有 is_generic=true 。

2.2.1. 案例

下面是一个简单的案例。

1设置Hive Metastore

首先需要有一个正在运行的 hive metastore。

然后，在本地文件 /opt/hive-conf/hive-site.xml 中配置以下参数，和运行的 hive 实例中的文件配置一致，可以直接使用 hive 实例的 hive-site.xml 文件。

<configuration>
   <property>
      <name>javax.jdo.option.ConnectionURL</name>
      <value>jdbc:mysql://localhost/metastore?createDatabaseIfNotExist=true</value>
      <description>metadata is stored in a MySQL server</description>
   </property>

   <property>
      <name>javax.jdo.option.ConnectionDriverName</name>
      <value>com.mysql.jdbc.Driver</value>
      <description>MySQL JDBC driver class</description>
   </property>

   <property>
      <name>javax.jdo.option.ConnectionUserName</name>
      <value>...</value>
      <description>user name for connecting to mysql server</description>
   </property>

   <property>
      <name>javax.jdo.option.ConnectionPassword</name>
      <value>...</value>
      <description>password for connecting to mysql server</description>
   </property>

   <property>
       <name>hive.metastore.uris</name>
       <value>thrift://localhost:9083</value>
       <description>IP address (or fully-qualified domain name) and port of the metastore host</description>
   </property>

   <property>
       <name>hive.metastore.schema.verification</name>
       <value>true</value>
   </property>

</configuration>

通过 hive 客户端连接 hive，然后运行一些命令，来测试连接是否可用。我们可以看到有一个数据库叫 default ，并且数据库中没有任何表。

hive> show databases;
OK
default
Time taken: 0.032 seconds, Fetched: 1 row(s)

hive> show tables;
OK
Time taken: 0.028 seconds, Fetched: 0 row(s)

2启动SQL CLI，使用Flink SQL DDL创建Hive catalog

将所有的 hive 依赖添加到 flink 分布式环境的 /lib 目录下，然后在 flink sql client 中创建 hive catalog

Flink SQL> CREATE CATALOG myhive WITH (
  'type' = 'hive',
  'hive-conf-dir' = '/opt/hive-conf'
);

3设置kafka集群

启动 kafka 集群，并且创建一个名叫 test 的主题，然后往该主题中写入一些二元组数据，包括姓名和年龄。

localhost$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>tom,15
>john,21

然后可以通过 kafka 客户端消费者看到这些消息。

localhost$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

tom,15
john,21

4启动 SQL 客户端，通过flink SQL DDL 语句创建 kafka 表

启动 Flink SQL 客户端，通过 DDL 创建一个简单的 kafka 表，然后检查他的 schema。

Flink SQL> USE CATALOG myhive;

Flink SQL> CREATE TABLE mykafka (name String, age Int) WITH (
   'connector.type' = 'kafka',
   'connector.version' = 'universal',
   'connector.topic' = 'test',
   'connector.properties.bootstrap.servers' = 'localhost:9092',
   'format.type' = 'csv',
   'update-mode' = 'append'
);
[INFO] Table has been created.

Flink SQL> DESCRIBE mykafka;
root
 |-- name: STRING
 |-- age: INT

也可以通过 hive 客户端检查这张表，并且注意表包含属性：is_generic=true。

hive> show tables;
OK
mykafka
Time taken: 0.038 seconds, Fetched: 1 row(s)

hive> describe formatted mykafka;
OK
# col_name            	data_type           	comment


# Detailed Table Information
Database:           	default
Owner:              	null
CreateTime:         	......
LastAccessTime:     	UNKNOWN
Retention:          	0
Location:           	......
Table Type:         	MANAGED_TABLE
Table Parameters:
	flink.connector.properties.bootstrap.servers	localhost:9092
	flink.connector.topic	test
	flink.connector.type	kafka
	flink.connector.version	universal
	flink.format.type   	csv
	flink.generic.table.schema.0.data-type	VARCHAR(2147483647)
	flink.generic.table.schema.0.name	name
	flink.generic.table.schema.1.data-type	INT
	flink.generic.table.schema.1.name	age
	flink.update-mode   	append
	is_generic          	true
	transient_lastDdlTime	......

# 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.158 seconds, Fetched: 36 row(s)

5通过 flink sql 查询 kafka 表

通过 flink sql 客户端运行一个简单的 select 语句到 flink 集群，不管是使用 standalone 还是 yarn-session。

Flink SQL> select * from mykafka;

往 kafka 主题中发送一些数据。

localhost$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

tom,15
john,21
kitty,30
amy,24
kaiky,18

然后就可以从 flink sql 客户端看到产生的结果。

             SQL Query Result (Table)
 Refresh: 1 s    Page: Last of 1     

        name                       age
         tom                        15
        john                        21
       kitty                        30
         amy                        24
       kaiky                        18

2.3. 支持的类型

HiveCatalog支持通用表的所有flink类型。

对于hive兼容表，HiveCatalog需要映射flink数据类型到hive类型，就像下表描述的一样：

对于类型映射，需要注意一些事情：

• hive的 CHAR§类型最大长度为255。
• hive的VARCHAR§类型最大长度为65535。
• hive的MAP key仅支持基本的类型，但是flink的MAP支持所有数据类型。
• hive的UNION类型目前不支持。
• 通常情况下，hive的TIMESTAMP类型精度为9，并且不支持其他精度。hive的UDF和其他的处理，可以处理精度 <=9 的TIMESTAMP值。
• hive不支持flink的TIMESTAMP_WITH_TIME_ZONE、TIMESTAMP_WITH_LOCAL_TIME_ZONE、MULTISET。
• 目前，flink的INTERVAL类型还不能映射到hive的INTERVAL类型。

3. 读写Hive

使用HiveCatalog，Flink可以使用统一的流或批模式来处理Hive表。这意味着flink可以比Hive的批引擎有更好的性能，而且还可以连续的读取或写入数据到Hive表来提供数据库实时处理能力。

3.1. 读

Flink支持从Hive中以批或流模式来读取数据。当使用批模式运行时，Flink将会对表当前时间点的数据进行查询。流式读取将会持续的监控表，并且抓取可见的新数据。Flink默认以有界流读取表。

流式读取支持消费分区表和无分区的表。对于分区表，Flink将会监控新生成的分区，并且在它们可见时马上读取。对于无分区的表，Flink将会监控目录下新生成的文件，并且读取它们。

SQL提示可以用于给hive表提供配置，而无需在hive元数据中更改表的定义。

SELECT * FROM hive_table /*+ 
    OPTIONS(
        'streaming-source.enable'='true', 
        'streaming-source.consume-start-offset'='2020-05-20'
    ) 
    */
;

注意：

• 监控策略会读取当前路径下的所有目录/文件，太多的分区可能导致性能下降。
• 流式读取无分区表要求每个文件被原子性地写入目标目录。
• 流式读取分区表要求每个分区被原子性的加入hive的元数据。否则，新增加到一个已存在的分区的数据将会被消费。
• 在Flink DDL中，流式读取不支持水印语法，这些表不能被用于窗口操作。

3.1.1. 读取hive视图

Flink支持读取Hive视图，但是会有以下限制：

• 在读取视图之前，hive catalog必须被设置为当前的catalog。可以通过USE CATALOG ...来设置。
• flink SQL和hive SQL有不同的语法，比如不同的关键字和字面量。请确保时区的查询和flink语法兼容。

3.1.2. 向量读优化

当符合下面的条件时，flink将对hive表自动使用向量读：

• 格式：ORC或Parquet。
• 非复杂数据类型，比如hive类型：List、Map、Struct、Union。

这个特性默认开启，可以通过下面的配置来禁用：

'table.exec.hive.fallback-mapred-reader'='true';

3.1.3. source并行度推断

默认情况下，flink会基于要读取的hive表对应的文件数量以及每个文件的文件块数量来推断最优的并行度。

flink允许配置并行度推断的策略，可以通过 sql 来修改下面的配置。注意：这些参数将会影响整个job中所有的source。

3.1.4. 调整读取hive表的数据分片大小

读 Hive 表时, 数据文件将会被切分为若干个分片（split）, 每一个分片是要读取的数据的一部分。 分片是 Flink 进行任务分配和数据并行读取的基本粒度。 用户可以通过下面的参数来调整每个分片的大小来做一定的读性能调优。

NOTE:

• 为了调整数据分片的大小， Flink 首先将计算得到所有分区下的所有文件的大小。 但是这在分区数量很多的情况下会比较耗时，你可以配置作业参数 table.exec.hive.calculate-partition-size.thread-num（默认为3）为一个更大的值使用更多的线程来进行加速。
• 目前上述参数仅适用于 ORC 格式的 Hive 表。

3.1.4.1. 读取带有子目录的分区

在某些情况下，你或许会创建一个引用其他表的外部表，但是该表的分区列是另一张表分区字段的子集。 比如，你创建了一个分区表 fact_tz，分区字段是 day 和 hour：

CREATE TABLE fact_tz(x int) PARTITIONED BY (day STRING, hour STRING);

然后你基于 fact_tz 表创建了一个外部表 fact_daily，并使用了一个粗粒度的分区字段 day：

CREATE EXTERNAL TABLE fact_daily(x int) PARTITIONED BY (ds STRING) LOCATION '/path/to/fact_tz';

当读取外部表 fact_daily 时，该表的分区目录下存在子目录(hour=1 到 hour=24)。

默认情况下，可以将带有子目录的分区添加到外部表中。Flink SQL 会递归扫描所有的子目录，并获取所有子目录中数据。

ALTER TABLE fact_daily ADD PARTITION (ds='2022-07-07') location '/path/to/fact_tz/ds=2022-07-07';

你可以设置作业属性 table.exec.hive.read-partition-with-subdirectory.enabled (默认为 true) 为 false 以禁止 Flink 读取子目录。 如果你设置成 false 并且分区目录下不包含任何子目录，Flink 会抛出 java.io.IOException: Not a file: /path/to/data/* 异常。

3.2. 时态表连接（temporal table join）

可以将hive表作为时态表，然后流就可以通过temporal join关联这个hive表。请查看temporal join来了解更多关于temporal join的信息。

flink支持使用处理时间来 temporal join hive表，处理时间 temporal join 通常会关联时态表的最新版本数据。
flink支持 temporal join 分区表和无分区表。对于分区表，flink支持自动跟踪hive表的最新分区。

注：flink目前不支持通过事件时间来 temporal join hive表。

3.2.1. Temporal Join最新分区

对于随时间变化的分区表，可以将其作为一个无界流来读取。如果每个分区包含一个版本的完整数据，分区就可以被作为时态表的一个版本。时态表的每个版本数据对应一个分区。

对于处理时间的 temporal join，flink会自动跟踪时态表最新的分区（版本）。最新分区（版本）通过 streaming-source.partition-order 选项定义。在flink流应用程序任务中，将hive表作为维表是最常见的案例。

注：这个特性只支持流模式。

下面展示一个经典的商业数据处理pipeline，维表数据来自于hive，并且底层表数据通过批处理pipeline或flink任务去每天更新一次。kafka流数据来自于实时的在线商业数据或日志，并且需要关联维表来丰富流数据。

-- 确保hive表的数据每天更新一次，每天包含最新且完整的维度数据
SET 'table.sql-dialect'='hive';
CREATE TABLE dimension_table (
    product_id STRING,
    product_name STRING,
    unit_price DECIMAL(10, 4),
    pv_count BIGINT,
    like_count BIGINT,
    comment_count BIGINT,
    update_time TIMESTAMP(3),
    update_user STRING,
    ...
)
PARTITIONED BY (pt_year STRING, pt_month STRING, pt_day STRING)
TBLPROPERTIES (
    -- 使用默认的partition-name顺序，每12小时加载一次最新分区数据（最推荐的且合适的方式）
    'streaming-source.enable' = 'true',
    'streaming-source.partition.include' = 'latest',
    'streaming-source.monitor-interval' = '12 h',
    'streaming-source.partition-order' = 'partition-name', -- 默认选项，可以忽略
    
    -- 每12小时，使用分区文件的创建时间create-time顺序加载一次最新分区
    'streaming-source.enable' = 'true',
    'streaming-source.partition.include' = 'latest',
    'streaming-source.partition-order' = 'create-time',
    'streaming-source.monitor-interval' = '12 h'
    
    -- 每12小时，使用分区时间partition-time顺序加载一次最新分区
    'streaming-source.enable' = 'true',
    'streaming-source.partition.include' = 'latest',
    'streaming-source.monitor-interval' = '12 h',
    'streaming-source.partition-order' = 'partition-time',
    'partition.time-extractor.kind' = 'default',
    'partition.time-extractor.timestamp-pattern' = '$pt_year-$pt_month-$pt_day 00:00:00'
);

SET 'table.sql-dialect'='default';
CREATE TABLE orders_table (
    order_id STRING,
    order_amount DOUBLE,
    product_id STRING,
    log_ts TIMESTAMP(3),
    proctime as PROCTIME()
) WITH (...);

-- 流模式sql，kafka时态连接temporal join一张hive维表，flink将根据'streaming-source.monitor-interval‘配置的时间间隔，从配置的最新分区自动加载数据。
SELECT * FROM orders_table AS o
JOIN dimension_table FOR SYSTEM_TIME AS OF o.proctime AS dim
ON o.product_id = dim.product_id;

3.2.2. temporal join最新表

对于hive表，可以将其作为有界流读取。在这种情况下，我们在某个时间点查询时，只能查询到hive表的最新版本数据。表的最新版本数据包含hive表的所有数据。

当执行关联最新hive表的查询时，hive表数据将会被缓存到slot的内存中，并且流中参与join的每条数据都会通过key来决定是否能找对对应的匹配值。
使用最新hive表作为时态表不要求任何其他的配置。作为可选项，可以通过以下属性来配置hive表数据缓存的TTL。缓存过期之后，hive表将会再次被scan以加载最新数据。

下面的实例展示加载hive表所有数据作为时态表：

-- 确保hive表中的数据会通过批处理pipeline每次以覆盖overwrite的形式写入
SET 'table.sql-dialect'='hive';
    CREATE TABLE dimension_table (
    product_id STRING,
    product_name STRING,
    unit_price DECIMAL(10, 4),
    pv_count BIGINT,
    like_count BIGINT,
    comment_count BIGINT,
    update_time TIMESTAMP(3),
    update_user STRING,
    ...
) TBLPROPERTIES (
    'streaming-source.enable' = 'false', -- 默认值，可以被忽略
    'streaming-source.partition.include' = 'all', -- 默认值，可以被忽略
    'lookup.join.cache.ttl' = '12 h'
);

SET 'table.sql-dialect'='default';
CREATE TABLE orders_table (
    order_id STRING,
    order_amount DOUBLE,
    product_id STRING,
    log_ts TIMESTAMP(3),
    proctime as PROCTIME()
) WITH (...);

-- 流式sql，kafka join hive维表。flink将会在缓存生存时间过期之后从dimension_table加载所有数据
SELECT * FROM orders_table AS o
JOIN dimension_table FOR SYSTEM_TIME AS OF o.proctime AS dim
ON o.product_id = dim.product_id;

注：

• 每个join的子任务都需要在自己的缓存中保持hive表的数据。请确保hive表的数据可以被缓存到TM任务的一个slot内存中，也就是说，每一个slot所分配的内存都能容纳hive表的所有数据。
• 建议对 streaming-source.monitor-interval（最新分区作为时态表）和 lookup.join.cache.ttl（所有分区作为时态表）设置一个很大的值，否则任务会频繁的更新和重加载数据，从而造成性能问题。
• 目前，flink是简单的加载整个hive表，而不管缓存是否需要被冲刷新。现在没有方法去比较新数据和旧数据的不同。

3.3. 写

flink支持通过BATCH和STREAMING模式来写入数据到hive表。当运行BATCH模式应用程序时，flink写入hive表的数据，只能在任务完成后才能被看到。BATCH写入支持追加和覆盖已存在的表数据。

-- INSERT INTO将会追加表或分区数据，并且完好无损的保存已存在的数据
Flink SQL> INSERT INTO mytable SELECT 'Tom', 25;
-- INSERT OVERWRITE将会覆盖表或分区中已存在的数据
Flink SQL> INSERT OVERWRITE mytable SELECT 'Tom', 25;

数据也可以被插入常规分区：

-- 写入静态分区，直接在表后面指定分区对应的值
Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type='type_1', my_date='2019-08-08') SELECT 'Tom', 25;
-- 写入动态分区，根据分区字段的值来判断每条数据写入的分区
Flink SQL> INSERT OVERWRITE myparttable SELECT 'Tom', 25, 'type_1', '2019-08-08';
-- 插入静态分区（my_type）和动态分区（my_date）
Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type='type_1') SELECT 'Tom', 25, '2019-08-08';

流式写入会持续的增加新数据到hive表，并且递增的提交数据以使他们可见。用户可以通过几个配置来控制何时/怎样触发提交。INSERT OVERWRITE 不支持流式写入。

下面的例子展示如何使用分区提交，并通过流式查询以及流式sink来从kafka读取数据并写入hive表，然后运行批查询来读取写入到hive的数据。

请查看streaming sink章节来获取完整可用的配置的列表。

SET 'table.sql-dialect'='hive';
CREATE TABLE hive_table (
    user_id STRING,
    order_amount DOUBLE
) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
    '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'
);

SET 'table.sql-dialect'='default';
CREATE TABLE kafka_table (
    user_id STRING,
    order_amount DOUBLE,
    log_ts TIMESTAMP(3),
    WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND -- 在TIMESTAMP列上定义水印
) WITH (...);

-- 流式sql，insert into到hive表
INSERT INTO TABLE hive_table
SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH')
FROM kafka_table;

-- batch sql，通过指定分区查询数据
SELECT * FROM hive_table WHERE dt='2020-05-20' and hr='12';

如果水印被定义在TIMESTAMP_LTZ字段上，并且使用partition-time来提交，则必须配置 sink.partition-commit.watermark-time-zone 指定会话的时区，否则分区将会晚几个小时提交。

SET table.sql-dialect=hive;
CREATE TABLE hive_table (
    user_id STRING,
    order_amount DOUBLE
) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
    'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
    'sink.partition-commit.trigger'='partition-time',
    'sink.partition-commit.delay'='1 h',
    'sink.partition-commit.watermark-time-zone'='Asia/Shanghai', -- 确保用户配置的时区为Asia/Shanghai
    'sink.partition-commit.policy.kind'='metastore,success-file'
);

SET table.sql-dialect=default;
CREATE TABLE kafka_table (
    user_id STRING,
    order_amount DOUBLE,
    ts BIGINT, -- 纪元毫秒值
    ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
    WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND -- 在TIMESTAMP_LTZ字段上定义水印
) WITH (...);

-- 流式sql，insert into到hive表
INSERT INTO TABLE hive_table
SELECT user_id, order_amount, DATE_FORMAT(ts_ltz, 'yyyy-MM-dd'), DATE_FORMAT(ts_ltz, 'HH')
FROM kafka_table;

-- batch sql，通过指定分区查询数据
SELECT * FROM hive_table WHERE dt='2020-05-20' and hr='12';

默认情况下，对于流式写入，flink只支持重命名提交者，这意味着S3文件系统不支持切好一次的流式写入。
可以通过设置下面的参数为false来恰好一次写入S3，这将会通知flink使用自己的写入器，但是这只支持写入parquet和orc文件类型。这个配置会影响任务中的所有sink。

3.3.1. 动态分区的写入

不同于静态分区的写入总是需要用户指定分区列的值，动态分区允许用户在写入数据的时候不指定分区列的值。 比如，有这样一个分区表：

CREATE TABLE fact_tz(x int) PARTITIONED BY (day STRING, hour STRING);

用户可以使用如下的 SQL 语句向该分区表写入数据：

INSERT INTO TABLE fact_tz PARTITION (day, hour) select 1, '2022-8-8', '14';

在该 SQL 语句中，用户没有指定分区列的值，这就是一个典型的动态分区写入的例子。

默认情况下, 如果是动态分区的写入, 在实际写入目标表之前，Flink 将额外对数据按照动态分区列进行排序。 这就意味着 sink 节点收到的数据都是按分区排序的，即首先收到一个分区的数据，然后收到另一个分区的数据，不同分区的数据不会混在一起。 这样 Hive sink 节点就可以一次只维护一个分区的 writer，否则，Hive sink 需要维护收到的数据对应的所有分区的 writer，如果分区的 writer 过多的话，则可能会导致内存溢出（OutOfMemory）异常。

为了避免额外的排序，你可以将作业的配置项 table.exec.hive.sink.sort-by-dynamic-partition.enable（默认是 true）设置为 false。 但是这种配置下，如之前所述，如果单个 sink 节点收到的动态分区数过多的话，则有可能会出现内存溢出的异常。

如果数据倾斜不严重的话，你可以在 SQL 语句中添加 DISTRIBUTED BY <partition_field> 将相同分区的数据分布到到相同的 sink 节点上来缓解单个 sink 节点的分区 writer 过多的问题。

此外，你也可以在 SQL 语句中添加 DISTRIBUTED BY <partition_field> 来达到将 table.exec.hive.sink.sort-by-dynamic-partition.enable 设置为 false 的效果。

注意：

• 该配置项 table.exec.hive.sink.sort-by-dynamic-partition.enable 只在批模式下生效。
• 目前，只有在 Flink 批模式下使用了 Hive 方言，才可以使用 DISTRIBUTED BY 和 SORTED BY。

3.3.2. 自动收集统计信息

在使用 Flink 写入 Hive 表的时候，Flink 将默认自动收集写入数据的统计信息然后将其提交至 Hive metastore 中。 但在某些情况下，你可能不想自动收集统计信息，因为收集这些统计信息可能会花费一定的时间。 为了避免 Flink 自动收集统计信息，你可以设置作业参数 table.exec.hive.sink.statistic-auto-gather.enable (默认是 true) 为 false。

如果写入的 Hive 表是以 Parquet 或者 ORC 格式存储的时候，numFiles/totalSize/numRows/rawDataSize 这些统计信息可以被 Flink 收集到。 否则, 只有 numFiles/totalSize 可以被收集到。

对于 Parquet 或者 ORC 格式的表，为了快速收集到统计信息 numRows/rawDataSize， Flink 只会读取文件的 footer。但是在文件数量很多的情况下，这可能也会比较耗时，你可以通过 设置作业参数 table.exec.hive.sink.statistic-auto-gather.thread-num（默认是 3）为一个更大的值来加快统计信息的收集。

注意：

• 只有批模式才支持自动收集统计信息，流模式目前还不支持自动收集统计信息。

3.4. 格式

下面的文件格式已经通过了flink和hive的兼容测试：

• Text
• CSV
• SequenceFile
• ORC
• Parquet

4. Hive函数

4.1. 通过HiveModule使用hive内置函数

HiveModule提供将hive的内置函数作为flink系统的内置函数的功能，可以通过flink SQL使用。
详细信息，请参考HiveModule。

java

String name            = "myhive";
String version         = "2.3.4";

tableEnv.loadModue(name, new HiveModule(version));

scala

val name            = "myhive"
val version         = "2.3.4"

tableEnv.loadModue(name, new HiveModule(version));

python

from pyflink.table.module import HiveModule

name = "myhive"
version = "2.3.4"

t_env.load_module(name, HiveModule(version))

sql client

LOAD MODULE hive WITH ('hive-version' = '2.3.4');

旧版本的 hive 内置函数有线程安全问题，用户需要注意。

4.2. hive自定义函数

用户可以在flink中使用已存在的hive自定义函数。

支持的UDF类型包括：

• UDF
• GenericUDF
• GenericUDTF
• UDAF
• GenericUDAFResolver2

对于查询计划器和执行，hive的UDF和GenericUDF会被自动翻译为flink的ScalarFunction，hive的GenericUDTF被自动翻译为flink的TableFunction，
hive的 UDAF 和 GenericUDAFResolver2 被翻译为flink的 AggregateFunction。

为了使用hive自定义函数 ，用户需要：

• 设置会话当前的catalog为HiveCatalog，来支持读取包含这些函数的hive元数据。
• 将包含函数的jar加载到flink的classpath。

4.3. 使用hive自定义函数

确保下面的hive函数被注册到hive的元数据：

/**
 * 测试简单的udf，注册函数名为：myudf。<br>
 * 已过期，建议使用GenericUDF
 */
public class TestHiveSimpleUDF extends UDF {

  public IntWritable evaluate(IntWritable i) {
    return new IntWritable(i.get());
  }

  public Text evaluate(Text text) {
    return new Text(text.toString());
  }
}

/**
 * 测试generic udf.注册函数名称为：mygenericudf
 */
public class TestHiveGenericUDF extends GenericUDF {

  @Override
  public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    checkArgument(arguments.length == 2);
    checkArgument(arguments[1] instanceof ConstantObjectInspector);
    Object constant = ((ConstantObjectInspector) arguments[1]).getWritableConstantValue();
    checkArgument(constant instanceof IntWritable);
    checkArgument(((IntWritable) constant).get() == 1);

    if (arguments[0] instanceof IntObjectInspector || arguments[0] instanceof StringObjectInspector) {
      return arguments[0];
    } else {
      throw new RuntimeException("Not support argument: " + arguments[0]);
    }
  }

  @Override
  public Object evaluate(DeferredObject[] arguments) throws HiveException {
    return arguments[0].get();
  }

  @Override
  public String getDisplayString(String[] children) {
    return "TestHiveGenericUDF";
  }
}

/**
 * 测试split udtf。注册函数名为：mygenericudtf
 */
public class TestHiveUDTF extends GenericUDTF {

  @Override
  public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
    checkArgument(argOIs.length == 2);
    //测试常量参数
    checkArgument(argOIs[1] instanceof ConstantObjectInspector);
    Object constant = ((ConstantObjectInspector) argOIs[1]).getWritableConstantValue();
    checkArgument(constant instanceof IntWritable);
    checkArgument(((IntWritable) constant).get() == 1);

    return ObjectInspectorFactory.getStandardStructObjectInspector(
            Collections.singletonList("col1"),
            Collections.singletonList(PrimitiveObjectInspectorFactory.javaStringObjectInspector));
  }

  @Override
  public void process(Object[] args) throws HiveException {
    String str = (String) args[0];
    for (String s : str.split(",")) {
      forward(s);
      forward(s);
    }
  }

  @Override
  public void close() {
  }
}

通过hive CLI客户端，我们可以看到他们应被注册，也就是说，应该在使用 flink sql 之前，就将他们注册为 hive 的系统函数：

hive> show functions;
OK
......
mygenericudf
myudf
myudtf

然后，用户就可以在flink SQL中使用他们了：文章来源地址https://www.toymoban.com/news/detail-593630.html

Flink SQL> select mygenericudf(myudf(name), 1) as a, mygenericudf(myudf(age), 1) as b, s from mysourcetable, lateral table(myudtf(name, 1)) as T(s);

到了这里，关于flink-sql读写hive-1.16的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！