Flink integrates Paimon

This article provides an implementation plan for Flink to integrate Paimon scenarios, including configuration examples, execution steps, and parameter descriptions.

Case

Use flink to add, delete, modify and query Paimon data

solution

Create Flink SQL query job and download 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

Flink SQL

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

-- 选择Catalog
USE CATALOG hive_catalog;

-- 选择数据源
USE ispong_db;

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

-- 插入数据
INSERT INTO users_paimon2 VALUES 
('zhangsan', 13),
('lisi', 14);