文档首页/ 智能数据湖 AI DataLake/ 开发指南/ Paimon开发指南/ Paimon on Flink/ 查询和写入Paimon表数据
更新时间：2026-10-16 GMT+08:00
查看PDF
分享

查询和写入Paimon表数据

注意事项

  • 需要底层文件系统（如OBS）可用。
  • 主键表必须定义PRIMARY KEY，且主键列不能为NULL。
  • 流式写入Paimon时需要开启checkpoint，数据在checkpoint时提交。
  • 主键表的分桶数（bucket）建议与Sink并行度一致，避免数据倾斜。
  • Paimon的分区键必须包含在主键中。
  • Flink Streaming作业不支持insert overwrite。
  • Paimon表使用Catalog方式管理元数据后，不允许rename。
  • Paimon不支持Sink materializer，请配置table.exec.sink.upsert-materialize=NONE。

代码示例

  • 示例1：无主键数据追加写入Paimon Append表。 
    CREATE CATALOG lf_paimon_catalog
WITH
  (
    'type' = 'paimon',
    'metastore' = 'lakeformation'
  );

CREATE DATABASE IF NOT EXISTS lf_paimon_catalog.paimon_db
WITH (
  'location' = 'obs://bucket/path/paimon-warehouse/paimon_db'
);

CREATE TABLE IF NOT EXISTS lf_paimon_catalog.paimon_db.paimon_table (
  id BIGINT,
  customer_id STRING,
  amount DOUBLE,
  region STRING
);

CREATE TABLE kafka_source(
  id BIGINT,
  customer_id STRING,
  amount DOUBLE,
  region STRING
)
WITH
  (
    'connector' = 'kafka',
    'topic' = 'user-events',
    'properties.bootstrap.servers' = '127.0.0.1:9092',
    'properties.group.id' = 'flink-consumer-group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
  );

INSERT INTO
  lf_paimon_catalog.paimon_db.paimon_table 
SELECT
  *
FROM
  kafka_source;
  • 示例2：读取Paimon Append表。 
    CREATE CATALOG lf_paimon_catalog
WITH
  (
    'type' = 'paimon',
    'metastore' = 'lakeformation'
  );

CREATE TABLE sink (
  id BIGINT,
  customer_id STRING,
  amount DOUBLE,
  region STRING
)
WITH
  ('connector' = 'print');

INSERT INTO
  sink
SELECT
  *
FROM
  lf_paimon_catalog.paimon_db.paimon_table /*+ OPTIONS('streaming'='true', 'monitor-interval'='60s') */;
父主题： Paimon on Flink

相关文档