Flink 1.18 实时计算：Kafka到Hive全流程

1. 环境准备

• Flink 1.18集群：需部署Standalone/Yarn/K8s模式
• Kafka：创建Topic（如user_behavior）
• Hive Metastore：配置Hive Catalog（需hive-exec和flink-connector-hive JAR包）
• 依赖JAR：

  flink-sql-connector-kafka-3.1.0-*.jar  # Kafka连接器
  flink-connector-hive_2.12-*.jar        # Hive连接器
  

2. 定义Kafka数据源表

CREATE TABLE kafka_source (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND  -- 水位线定义
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'kafka-broker:9092',
    'properties.group.id' = 'flink-group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

3. 定义Hive目标表

CREATE CATALOG hive_catalog WITH (
    'type' = 'hive',
    'hive-conf-dir' = '/path/to/hive-conf'
);

USE CATALOG hive_catalog;

CREATE TABLE IF NOT EXISTS hive_sink (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    event_time TIMESTAMP,
    dt STRING   -- 分区字段
) PARTITIONED BY (dt) STORED AS ORC TBLPROPERTIES (
    'sink.partition-commit.delay' = '1 h',
    'sink.partition-commit.policy.kind' = 'metastore,success-file'
);

4. 实时ETL处理逻辑

INSERT INTO hive_sink 
SELECT 
    user_id,
    item_id,
    behavior,
    event_time,
    DATE_FORMAT(event_time, 'yyyy-MM-dd') AS dt  -- 动态分区
FROM kafka_source
WHERE behavior = 'purchase';  -- 过滤有效行为

5. 作业提交与监控

• 提交作业：

  ./bin/sql-client.sh -f /path/to/etl_job.sql
  

• 关键配置：

  SET 'execution.checkpointing.interval' = '1min';
  SET 'table.exec.hive.infer-source-parallelism' = 'false';
  

• 监控：
  • Flink Web UI：检查Checkpoints状态
  • Hive Metastore：验证分区生成
  • HDFS：检查数据文件（/warehouse/hive_sink/dt=2024-01-01）

6. 技术要点

1. 动态分区：通过DATE_FORMAT自动生成分区路径
2. 流式写入：Flink自动管理分区提交（需配置sink.partition-commit.*）
3. 格式兼容：ORC/Parquet格式需保证Schema一致性
4. 水位线机制：确保事件时间语义正确性
5. 连接器优化：

   'properties.auto.offset.reset' = 'earliest'  -- 从最早偏移量消费
   'sink.batch-flush.max-size' = '128MB'        -- 批次写入优化
   

故障处理：若作业失败，Flink Checkpoint机制可恢复至最近一致状态，Kafka偏移量自动回滚。

7. 验证数据

SELECT dt, COUNT(*) FROM hive_sink GROUP BY dt;  -- 在Hive中查询分区数据量