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
的依赖项：
exportHADOOP_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
目录中。
hive版本maven依赖2.3.0 - 2.3.9flink-sql-connector-hive-2.3.93.0.0 - 3.1.2flink-sql-connector-hive-3.1.2
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 =newHiveCatalog(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 为会话的当前 catalogcurrent-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 为会话的当前 catalogUSE CATALOG myhive;
下面是使用 YAML 文件或 DDL 语句创建 HiveCatalog 实例时支持的一些参数。
参数要求默认值类型描述typeYes(none)Stringcatalog 的类型，创建 HiveCatalog 时必须设置为：hivenameYes(none)Stringcatalog 的唯一名称，只适用于 YAML 文件。hive-conf-dirNo(none)String包含 hive-site.xml 文件的 hive 目录的 URI，而且必须是 hadoop 文件系统支持的 URI。如果使用了相对路径的 URI ，比如没有指定 schema ，则会假定指定的本地文件系统。如果没有指定该选项，flink 将会从 class path 中搜索 hive-site.xml 文件。default-databaseNodefaultString指定当前 catalog 使用的默认数据库。hive-versionNo(none)StringHiveCatalog 有能力自动推断使用的 hive 版本，建议不要指定 hive 版本号，除非自动推断失败。hadoop-conf-dirNo(none)Stringhadoop 配置文件路径，只支持本地文件系统路径。建议的指定 hadoop 配置文件目录的方式是将 HADOOP_CONF_DIR 配置为环境变量，比如 /etc/profile。只有在环境变量不起作用时，使用该参数，比如你想分别配置每个 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>showdatabases;
OK
defaultTime taken: 0.032 seconds, Fetched: 1row(s)
hive>showtables;
OK
Time taken: 0.028 seconds, Fetched: 0row(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>CREATETABLE 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>showtables;
OK
mykafka
Time taken: 0.038 seconds, Fetched: 1row(s)
hive>describe formatted mykafka;
OK
# col_name data_type comment# Detailed Table InformationDatabase: default
Owner: null
CreateTime: ......
LastAccessTime: UNKNOWN
Retention: 0
Location: ......TableType: 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-typeVARCHAR(2147483647)
flink.generic.table.schema.0.name name
flink.generic.table.schema.1.data-typeINT
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 1Time taken: 0.158 seconds, Fetched: 36row(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类型，就像下表描述的一样：
Flink 数据类型Hive 数据类型
CHAR(p)
CHAR(p)
VARCHAR(p)
VARCHAR(p)
STRINGSTRINGBOOLEANBOOLEANTINYINTTINYINTSMALLINTSMALLINTINTINTBIGINTLONGFLOATFLOATDOUBLEDOUBLEDECIMAL(p, s)DECIMAL(p, s)DATEDATETIMESTAMP(9)TIMESTAMPBYTESBINARYARRAYLISTMAPMAPROWSTRUCT
对于类型映射，需要注意一些事情：
- 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将会监控目录下新生成的文件，并且读取它们。
Key默认值类型描述streaming-source.enablefalseBoolean是否开启流式source。注意：请确保每个分区/文件被原子性地写入，否则读取器可能会获取到不完整的数据。streaming-source.partition.includeallString设置读取分区的选项，可用选项为：‘all’和’latest’。
'all’意味着读取所有分区；
‘latest’ 意味着按照’
streaming-source.partition.order
’设置的分区顺序来读取最新的分区。
‘latest’只会在流式source的hive表作为时态表时起作用。
默认值为’all’。
在开启
streaming-source.enable
，并且设置
streaming-source.partition.include
为
latest
时，Flink支持
temporal连接
最新的hive分区数据，同时，用户可以通过配置下面的分区相关的选项来指定分区的比较顺序和数据的更新间隔。streaming-source.monitor-intervalNoneDuration持续监控分区/文件的时间间隔。
注意：对于hive流式读取，默认值为’1 min’；对于hive流式temporal join，默认值为’60 min’，这是因为这儿有一个框架限制。
在当前的hive流式
temporal join
实现上，TM访问hive的元数据可能会对
hive metaStore
产生很大的压力，这将会在未来改进。streaming-source.partition-orderpartition-nameString流式source模式下的分区顺序，支持’create-time’、‘partition-time’、‘partition-name’。
'create-time’会比较分区/文件的创建时间，这并不是分区在hive元数据中的创建时间，而是目录/文件在文件系统中的修改时间，如果分区目录被更新了，比如往目录中增加了新的文件，这可能对数据消费造成影响。
‘partition-time’比较通过分区名称提取的时间。
‘partition-name’比较分区的字符串顺序。对于无分区表，这个值必须是’create-time’。
默认情况下，该值为’partition-name’。该选项和过期的
streaming-source.consume-order
作用一样。streaming-source.consume-start-offsetNoneString流式消费的起始偏移量。怎么转换和比较取决于你的设置。
对于
create-time
和
partition-time
，应该是一个时间戳格式化字符串（**yyyy-[m]m-[d]d [hh:mm:ss]**）。对于
partition-name
，将会使用分区时间提取器从分区中提取时间，并且分区名称应该有时间对应的值，比如：pt_year=2020/pt_mon=10/pt_day=01。
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。
Key默认值类型描述table.exec.hive.infer-source-parallelismtrueBoolean如果为true，将会通过文件块数量来推断source并行度。
如果为false，source的并行度将会通过配置来设置。table.exec.hive.infer-source-parallelism.max1000Integer设置source算子的最大并行度。
3.1.4. 调整读取hive表的数据分片大小
读 Hive 表时, 数据文件将会被切分为若干个分片（split）, 每一个分片是要读取的数据的一部分。 分片是 Flink 进行任务分配和数据并行读取的基本粒度。 用户可以通过下面的参数来调整每个分片的大小来做一定的读性能调优。
配置默认值类型描述table.exec.hive.split-max-size128mbMemorySize读 Hive 表时，每个分片最大可以包含的字节数 (默认是 128MB)table.exec.hive.file-open-cost4mbMemorySize打开一个文件预估的开销，以字节为单位，默认是 4MB。 如果这个值比较大，Flink 则将会倾向于将 Hive 表切分为更少的分片，这在 Hive 表中包含大量小文件的时候很有用。 反之，Flink 将会倾向于将 Hive 表切分为更多的分片，这有利于提升数据读取的并行度。
NOTE:
- 为了调整数据分片的大小， Flink 首先将计算得到所有分区下的所有文件的大小。 但是这在分区数量很多的情况下会比较耗时，你可以配置作业参数
table.exec.hive.calculate-partition-size.thread-num（默认为3）为一个更大的值使用更多的线程来进行加速。 - 目前上述参数仅适用于 ORC 格式的 Hive 表。
3.1.4.1. 读取带有子目录的分区
在某些情况下，你或许会创建一个引用其他表的外部表，但是该表的分区列是另一张表分区字段的子集。 比如，你创建了一个分区表
fact_tz
，分区字段是
day
和
hour
：
CREATETABLE 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 会递归扫描所有的子目录，并获取所有子目录中数据。
ALTERTABLE fact_daily ADDPARTITION(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';CREATETABLE 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';CREATETABLE 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 ASOF 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以加载最新数据。
Key默认值类型描述lookup.join.cache.ttl60 minDurationlookup join时缓存的TTL（比如 10min）。默认值值是60分钟。
注：这个选项只在lookup有界hive表source时起作用，如果使用流hive source作为时态表，请使用
streaming-source.monitor-interval
配置数据更新的时间。
下面的实例展示加载hive表所有数据作为时态表：
-- 确保hive表中的数据会通过批处理pipeline每次以覆盖overwrite的形式写入SET'table.sql-dialect'='hive';CREATETABLE 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';CREATETABLE 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 ASOF 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>INSERTINTO 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';CREATETABLE 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';CREATETABLE 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表INSERTINTOTABLE 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
指定会话的时区，否则分区将会晚几个小时提交。
SETtable.sql-dialect=hive;CREATETABLE 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');SETtable.sql-dialect=default;CREATETABLE 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表INSERTINTOTABLE 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。
Key默认值类型描述table.exec.hive.fallback-mapred-writertrueBoolean如果该设置为 false，将使用 flink 的 native write 写入数据到 parquet 和 orc 文件；
如果该设置为true，将使用 hadoop mapred 的 record writer 写入数据到 parquet 和 orc 文件。
3.3.1. 动态分区的写入
不同于静态分区的写入总是需要用户指定分区列的值，动态分区允许用户在写入数据的时候不指定分区列的值。 比如，有这样一个分区表：
CREATETABLE fact_tz(x int) PARTITIONED BY(day STRING,hour STRING);
用户可以使用如下的 SQL 语句向该分区表写入数据：
INSERTINTOTABLE fact_tz PARTITION(day,hour)select1,'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内置函数
HiveModul
e提供将hive的内置函数作为flink系统的内置函数的功能，可以通过flink SQL使用。
详细信息，请参考HiveModule。
java
String name ="myhive";String version ="2.3.4";
tableEnv.loadModue(name,newHiveModule(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
*/publicclassTestHiveSimpleUDFextends UDF {publicIntWritableevaluate(IntWritable i){returnnewIntWritable(i.get());}publicTextevaluate(Text text){returnnewText(text.toString());}}/**
* 测试generic udf.注册函数名称为：mygenericudf
*/publicclassTestHiveGenericUDFextendsGenericUDF{@OverridepublicObjectInspectorinitialize(ObjectInspector[] arguments)throwsUDFArgumentException{checkArgument(arguments.length ==2);checkArgument(arguments[1]instanceofConstantObjectInspector);Object constant =((ConstantObjectInspector) arguments[1]).getWritableConstantValue();checkArgument(constant instanceofIntWritable);checkArgument(((IntWritable) constant).get()==1);if(arguments[0]instanceofIntObjectInspector|| arguments[0]instanceofStringObjectInspector){return arguments[0];}else{thrownewRuntimeException("Not support argument: "+ arguments[0]);}}@OverridepublicObjectevaluate(DeferredObject[] arguments)throwsHiveException{return arguments[0].get();}@OverridepublicStringgetDisplayString(String[] children){return"TestHiveGenericUDF";}}/**
* 测试split udtf。注册函数名为：mygenericudtf
*/publicclassTestHiveUDTFextendsGenericUDTF{@OverridepublicStructObjectInspectorinitialize(ObjectInspector[] argOIs)throwsUDFArgumentException{checkArgument(argOIs.length ==2);//测试常量参数checkArgument(argOIs[1]instanceofConstantObjectInspector);Object constant =((ConstantObjectInspector) argOIs[1]).getWritableConstantValue();checkArgument(constant instanceofIntWritable);checkArgument(((IntWritable) constant).get()==1);returnObjectInspectorFactory.getStandardStructObjectInspector(Collections.singletonList("col1"),Collections.singletonList(PrimitiveObjectInspectorFactory.javaStringObjectInspector));}@Overridepublicvoidprocess(Object[] args)throwsHiveException{String str =(String) args[0];for(String s : str.split(",")){forward(s);forward(s);}}@Overridepublicvoidclose(){}}
通过hive CLI客户端，我们可以看到他们应被注册，也就是说，应该在使用 flink sql 之前，就将他们注册为 hive 的系统函数：
hive>show functions;
OK
......
mygenericudf
myudf
myudtf
然后，用户就可以在flink SQL中使用他们了：
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);
版权归原作者 第一片心意 所有， 如有侵权，请联系我们删除。