Hive的主要缺点在于查询的延迟很高，几乎成了离线分析的代言人。而 Flink 的特点就是实时性强，所以 Flink SQL 与 Hive 的结合势在必行。
Flink 与 Hive 的集成比较特别。Flink 提供了“Hive 目录”（HiveCatalog）功能，允许使用Hive 的“元存储”（Metastore）来管理 Flink 的元数据。这带来的好处体现在两个方面：
1）Metastore 可以作为一个持久化的目录，因此使用 HiveCatalog 可以跨会话存储 Flink特定的元数据。这样一来，我们在 HiveCatalog 中执行执行创建 Kafka 表或者 ElasticSearch 表，就可以把它们的元数据持久化存储在 Hive 的 Metastore 中；对于不同的作业会话就不需要重复创建了，直接在 SQL 查询中重用就可以。
2）使用 HiveCatalog，Flink 可以作为读写 Hive 表的替代分析引擎。这样一来，在 Hive中进行批处理会更加高效；与此同时，也有了连续在 Hive 中读写数据、进行流处理的能力，这也使得“实时数仓”（real-time data warehouse）成为了可能。
HiveCatalog 被设计为“开箱即用”，与现有的 Hive 配置完全兼容，我们不需要做任何的修改与调整就可以直接使用。注意只有 Blink 的计划器（planner）提供了 Hive 集成的支持，所以需要在使用 Flink SQL时选择Blink planner。
引入依赖
Hive 各版本特性变化比较大，所以使用时需要注意版本的兼容性。目前 Flink 支持的 Hive版本包括：
- Hive 1.x：1.0.0 ~ 1.2.2；
- Hive 2.x：2.0.0 ~ 2.2.0，2.3.0 ~ 2.3.6；
- Hive 3.x：3.0.0 ~ 3.1.2；
目前 Flink 与 Hive 的集成程度在持续加强，支持的版本信息也会不停变化和调整，可以随着关注官网的更新信息。
由于 Hive 是基于 Hadoop 的组件，因此我们首先需要提供 Hadoop 的相关支持，在环境变量中设置 HADOOP_CLASSPATH：
export HADOOP_CLASSPATH=`hadoop classpath`
在 Flink 程序中可以引入以下依赖：
<!-- Flink 的 Hive 连接器-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- Hive 依赖 -->
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>${hive.version}</version>
</dependency>
建议不要把这些依赖打包到结果 jar 文件中，而是在运行时的集群环境中为不同的 Hive版本添加不同的依赖支持。具体版本对应的依赖关系，可以查询官网说明。
连接到 Hive
在 Flink 中连接 Hive，是通过在表环境中配置 HiveCatalog 来实现的。需要说明的是，配置 HiveCatalog 本身并不需要限定使用哪个 planner，不过对 Hive 表的读写操作只有 Blink 的planner 才支持。所以一般我们需要将表环境的 planner 设置为 Blink。
下面是代码中配置 Catalog 的示例：
public class Udf_Hive {
public static void main(String[] args) {
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.inStreamingMode()
.useBlinkPlanner()
.build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
String name = "myhive";
String defaultDatabase = "mydatabase";
String hiveConfDir = "/opt/hive-conf";
// 创建一个 HiveCatalog，并在表环境中注册
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog("myhive", hive);
// 使用 HiveCatalog 作为当前会话的 catalog
tableEnv.useCatalog("myhive");
}
}
当然，我们也可以直接启动 SQL 客户端，用 CREATE CATALOG 语句直接创建 HiveCatalog：
Flink SQL> create catalog myhive with ('type' = 'hive', 'hive-conf-dir' = '/opt/hive-conf');
Flink SQL> use catalog myhive;
设置 SQL 方言
Hive内部提供了类SQL的查询语言，不过语法细节与标准SQL会有一些出入，相当于是 SQL 的一种“方言”（dialect）。为了提高与 Hive 集成时的兼容性，Flink SQL 提供了一个非常有趣而强大的功能：可以使用方言来编写 SQL 语句。换句话说，我们可以直接在 Flink中写 Hive SQL 来操作 Hive 表，这给我们的读写处理带来了极大的方便。
Flink 目前支持两种 SQL 方言的配置：default 和 hive。所谓的 default 就是 Flink SQL 默认的 SQL 语法了。我们需要先切换到 hive 方言，然后才能使用 Hive SQL 的语法。具体设置可以分为 SQL 和 Table API 两种方式。
1）SQL 中设置
我们可以通过配置 table.sql-dialect 属性来设置 SQL 方言：
set table.sql-dialect=hive;
当然，我们可以在代码中执行上面的 SET 语句，也可以直接启动 SQL 客户端来运行。如果使用 SQL 客户端，我们还可以在配置文件 sql-cli-defaults.yaml 中通过“configuration”模块来设置：
execution:
planner: blink
type: batch
result-mode: table
configuration:
table.sql-dialect: hive
2）Table API 中设置
另外一种方式就是在代码中，直接使用 Table API 获取表环境的配置项来进行设置：
// 配置 hive 方言
tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
// 配置 default 方言
tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
读写 Hive 表
有了 SQL 方言的设置，我们就可以很方便的在 Flink 中创建 Hive 表并进行读写操作了。Flink 支持以批处理和流处理模式向 Hive 中读写数据。在批处理模式下，Flink 会在执行查询语句时对 Hive 表进行一次性读取，在作业完成时将结果数据向 Hive 表进行一次性写入；而在流处理模式下，Flink 会持续监控 Hive 表，在新数据可用时增量读取，也可以持续写入新数据并增量式地让它们可见。更灵活的是，我们可以随时切换 SQL 方言，从其它数据源（例如 Kafka）读取数据、经转换后再写入Hive。我们可以启动SQL客户端来运行：
-- 设置 SQL 方言为 hive，创建 Hive 表
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' = '1h',
'sink.partition-commit.policy.kind' = 'metastore,success-file'
);
-- 设置 SQL 方言为 default，创建 Kafka 表
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 --定义水位线
) WITH (...);
-- 将 Kafka 中读取的数据经转换后写入 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;
创建 Hive 表时设置了通过分区时间来触发提交的策略。将 Kafka 中读取的数据经转换后写入 Hive，这是一个流处理的 Flink SQL 程序。
总结：Table API 和 SQL 是 Flink 最上层的应用接口，目前尚不完善，但发展非常迅速，每个小版本都会有底层优化和功能扩展。可以想到不久的将来，Flink SQL 将会是最为高效、最为普遍的开发手段，应该时刻保持跟进，随着框架的发展完善不断提升自己的技术能力。
