Hive 与 Flink 集成：实时读写 Hive 数据实战指南

1. 环境准备

• Hive 要求：Hive 2.3+（推荐 3.1.2），开启 Hive Metastore 服务
• Flink 要求：Flink 1.11+（推荐 1.14），启用 Hive Connector
• 依赖配置：在 Flink 的 lib 目录添加：

  flink-sql-connector-hive-3.1.2_2.12-1.14.3.jar
  hive-exec-3.1.2.jar
  

---

2. Hive Catalog 配置

通过 Catalog 连接 Hive Metastore：

// Flink SQL 环境初始化
EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);

// 创建 Hive Catalog
String catalogName = "hive_catalog";
HiveCatalog catalog = new HiveCatalog(
    catalogName,
    "default", // database
    "/path/to/hive-conf", // Hive 配置文件目录
    "2.3.7"    // Hive 版本
);

// 注册 Catalog
tableEnv.registerCatalog(catalogName, catalog);
tableEnv.useCatalog(catalogName);

---

3. 实时读取 Hive 表

3.1 流式读取

-- 创建 Hive 表映射
CREATE TABLE hive_source (
    user_id STRING,
    event_time TIMESTAMP(3)
) PARTITIONED BY (dt STRING) 
WITH (
    'connector' = 'hive',
    'hive-version' = '3.1.2'
);

-- 流式查询
SELECT user_id, COUNT(*) 
FROM hive_source 
WHERE dt = '2023-07-20'
GROUP BY TUMBLE(event_time, INTERVAL '5' MINUTE), user_id;

3.2 分区监控

启用动态分区发现：

ALTER TABLE hive_source SET ('streaming-source.enable' = 'true');

---

4. 实时写入 Hive 表

4.1 写入静态分区

-- 创建 Sink 表
CREATE TABLE hive_sink (
    product_id STRING,
    sales_cnt BIGINT
) PARTITIONED BY (dt STRING) 
WITH (
    'connector' = 'hive',
    'sink.partition-commit.delay' = '1 h'
);

-- 写入数据
INSERT INTO hive_sink 
SELECT product_id, COUNT(*), '2023-07-20' 
FROM kafka_source 
GROUP BY product_id;

4.2 动态分区写入

SET table.exec.hive.sink.partition-commit.policy.kind = 'metastore';
INSERT INTO hive_sink 
SELECT product_id, sales_cnt, dt  -- 自动识别 dt 为分区字段
FROM realtime_sales_stream;

---

5. 高级配置

配置项	说明
streaming-source.partition.include	监听分区格式（如 latest 或 2023-07-*）
sink.partition-commit.trigger	分区提交策略（process-time 或 partition-time）
sink.partition-commit.policy.kind	提交方式（metastore/success-file）

---

6. 调优技巧

1. 并行度优化：

   // 设置写入并行度
   tableEnv.getConfig().set("table.exec.resource.default-parallelism", "4");
   

2. 小文件合并：

   ALTER TABLE hive_sink SET ('auto-compaction' = 'true');
   

3. 批流统一：

   SET table.exec.hive.infer-source-parallelism.max = '100';
   

---

7. 常见问题解决

• 问题1：写入时分区未更新
  方案：检查 sink.partition-commit.delay 是否小于数据延迟
• 问题2：Metastore 连接失败
  方案：确认 hive.metastore.uris 在配置文件中正确配置
• 问题3：数据类型不匹配
  方案：使用 Flink 的 CAST 函数转换类型，例如 CAST(ts AS TIMESTAMP(3))

提示：完整示例代码见 Flink Hive Connector 官方文档