Postgres实时Hive

本文提供Postgres实时Hive场景的实施方案，包括配置示例、执行步骤和参数说明。

案例

基于Paimon将Postgres实时到Hive，支持数据实时增删改查。

前提

1. 参考PostgreSQL实时同步文档，下载依赖并上传。

注意

hive client访问paimon数据需要下载paimon-hive-connector包，不需要重启hive服务，重启hive会话即可。

• paimon-hive-connector-3.1-0.9.0.jar下载

mv paimon-hive-connector-3.1-0.9.0.jar /opt/hive/lib/

方案

创建Flink SQL作业类型，下载PostgreSQL相关额外依赖包和以下依赖

• paimon-flink-1.18-1.4-20251105.003055-51.jar下载
• hive-Standalone-metastore-3.1.3.jar下载
• hive-common-3.1.3.jar下载
• hadoop-mapreduce-client-core-3.3.5.jar下载
• commons-lang-2.6.jar下载
• hive-storage-api-4.0.0.jar下载
• libthrift-0.9.3.jar下载
• libfb303-0.9.3.jar下载
• hive-exec-3.1.3.jar下载

手动插入数据

-- 创建Catalog
CREATE CATALOG hive_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs://106.15.56.126:9000/user/hive/warehouse',
    'metastore' = 'hive',
    'uri' = 'thrift://106.15.56.126:9083',
    'hive-conf-dir' = '/opt/hive/conf' 
);

-- 选择Catalog
USE CATALOG hive_catalog;

-- 选择数据源
USE ispong_db;

-- 创建Paimon表
CREATE TABLE IF NOT EXISTS users_paimon (
    username STRING,
    age INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
    'bucket' = '2',
    'bucket-key' = 'username',
    'changelog-producer' = 'lookup'
);

-- 插入数据
INSERT INTO users_paimon VALUES 
('user1', 25),
('user2', 30);

PostgreSQL实时同步数据到hive

注意：一定要开启checkpoint，否则可能无法写入数据 需要修改flinkConfig高级配置如下：

{
  "execution.checkpointing.interval": "10s",
  "execution.runtime-mode": "streaming",
  "jobmanager.memory.process.size": "1g",
  "parallelism.default": "1",
  "pipeline.name": "Postgres CDC to Paimon",
  "taskmanager.memory.process.size": "2g",
  "taskmanager.numberOfTaskSlots": "1"
}

-- 创建 Hive Catalog
CREATE CATALOG hive_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs://106.15.56.126:9000/user/hive/warehouse',
    'metastore' = 'hive',
    'uri' = 'thrift://106.15.56.126:9083',
    'hive-conf-dir' = '/opt/hive/conf'
);

-- 选择 Hive Catalog
USE CATALOG hive_catalog;

-- 选择数据库
USE ispong_db;

-- 创建 PostgreSQL CDC 源表
CREATE TEMPORARY TABLE postgres_source (
    username STRING,
    age INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    'hostname' = '106.15.56.126',
    'port' = '54302',
    'username' = 'root',
    'password' = 'root123',
    'database-name' = 'ispong_db',
    'schema-name' = 'public',
    'table-name' = 'users',
    'slot.name' = 'flink',
    'scan.incremental.snapshot.enabled' = 'true'
);

-- 创建 Paimon 表
CREATE TABLE IF NOT EXISTS users_paimon (
    username STRING,
    age INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
    'bucket' = '2',
    'bucket-key' = 'username',
    'changelog-producer' = 'input',
    'changelog-producer.compaction-interval' = '10s'
);

-- 执行数据同步
INSERT INTO users_paimon 
SELECT username, age FROM postgres_source