Postgres real-time Hive

This article provides an implementation plan for Postgres real-time Hive scenario, including configuration examples, execution steps, and parameter descriptions.

Case

Based on Paimon, Postgres is transferred to Hive in real time, supporting real-time addition, deletion, modification and query of data.

Premise

1. Refer to the PostgreSQL real-time synchronization documentation to download the dependencies and upload them.

Note

The hive client needs to download the paimon-hive-connector package to access paimon data. There is no need to restart the hive service. Just restart the hive session.

• paimon-hive-connector-3.1-0.9.0.jar download

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

Plan

Create a Flink SQL job type, download PostgreSQL related additional dependency packages and the following dependencies

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

Manually insert data

-- 创建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 synchronizes data to hive in real time

Note: Checkpoint must be turned on, otherwise data may not be written. The flinkConfig advanced configuration needs to be modified as follows:

{
  "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