AI 助理
文档备案控制台
官方文档
首页 实时计算 Flink版 流计算Flink 开发参考 连接器 流式数据湖仓Paimon Paimon数据摄入连接器

Paimon数据摄入连接器

更新时间：
复制 MD 格式
产品详情
我的收藏

本文为您介绍如何在数据摄入 YAML 作业中，使用 Paimon 连接器作为目标端进行数据同步。

特色功能

  • 自动建表

    写入前无需在Paimon中预先创建目标表，数据摄入作业会根据上游的表结构自动创建对应的Paimon表。如果目标表已存在，则会校验已有表结构的兼容性并自动适配。

  • 表结构变更同步

    通过Pipeline级别的配置项schema.change.behavior控制表结构变更策略，将上游的表结构变更实时应用到Paimon表。详情请参见表结构变更。

  • 整库同步与分库分表合并

    支持单表同步、整库同步、整库同步并替换库表名、分库分表合并、整库同步时排除部分表等场景，详情请参见使用示例。

  • 复用已有Catalog

    自VVR 11.5版本起，可以直接引用“数据管理”页面中创建的内置Paimon Catalog，减少手写连接属性的工作量。详情请参见复用已有 Catalog。

使用限制

  • 数据摄入YAML作业仅支持将Paimon作为目标端，不支持作为数据源。

  • Paimon与VVR版本对应关系详情如下表所示。

    Paimon社区版本

    实时计算Flink版引擎版本（VVR ）

    2.0.0

    11.9

    1.3.1

    11.5、11.6、11.7、11.8

    1.3

    11.4

    1.2

    11.2、11.3

    1.1

    11.1

    1.0

    8.0.11

  • 部分配置项对实时计算引擎VVR版本有要求，详情请参见配置项表格的备注列。

注意事项

  • 并发写入场景的存储建议

    当多个作业并发更新同一张 Paimon 表时，标准 OSS 存储（oss://）可能因文件操作的原子性限制，导致小概率的提交冲突或作业报错。

    为确保持续稳定的写入，建议使用具备强原子性保障的元数据或存储服务。推荐优先使用数据湖构建（DLF），能提供Paimon元数据与存储服务统一管理，也可选择 OSS-HDFS 或 HDFS。

  • 参数配置的生效机制

    修改Paimon表的配置参数后，必须重启相关作业，新配置才能生效。运行中的作业无法动态感知或加载此类参数变更。

数据摄入

Paimon连接器可以用于数据摄入YAML作业开发，作为目标端写入。

语法结构

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

配置项

参数

说明

是否必填

数据类型

默认值

备注

type

连接器类型。

是

STRING

无

固定值为paimon。

name

目标端名称。

否

STRING

无

Sink的名称。

catalog.properties.metastore

Paimon Catalog的类型。

否

STRING

filesystem

取值如下：

  • filesystem（默认值）

  • rest（仅支持DLF，不支持DLF-Legacy）

catalog.properties.*

创建Paimon Catalog的参数。

否

STRING

无

详情请参见管理Paimon Catalog。

table.properties.*

创建Paimon table的参数。

否

STRING

无

详情请参见Paimon table options。

table.properties.file.format

表中数据文件的存储类型。

String

否

parquet

可选值：

  • orc

  • parquet

  • avro

  • lance (VVR 11.6及以上版本）

  • mosaic (VVR 11.9及以上版本)

catalog.properties.warehouse

文件存储的根目录。

否

STRING

无

仅在catalog.properties.metastore设置为 filesystem时生效。

commit.user-prefix

提交数据文件时的用户名前缀。

否

STRING

无

说明

建议为不同的作业设置不同的用户名，方便在出现提交冲突时定位冲突的作业。

partition.key

每个分区表的分区字段。

否

STRING

无

不同的表使用;分割，不同的字段使用,分割，表与字段使用:分割。例如：testdb.table1:id1,id2;testdb.table2:name。

sink.cross-partition-upsert.tables

指定需要跨分区更新（主键不包含所有分区字段）的表。

否

STRING

无

适用于跨分区更新的表。

  • 格式：使用分号;分隔表名。

  • 性能建议：此操作消耗大量计算资源。请为这些表创建独立的作业。

重要

  • 必须列出所有符合条件的表。 遗漏表名将导致数据重复。

sink.commit.parallelism

指定Commit算子的并发度。

否

INTEGER

无

在Commit算子成为瓶颈时可以通过该参数指定Commit算子的并发度以提升性能。

仅实时计算引擎VVR 11.6及以上版本支持该参数。

说明

设置该参数会导致算子并发度修改，有状态重启作业时需要指定允许忽略部分算子状态（AllowNonRestoredState）。

sink.existing-table.schema-check.enabled

写入预先创建的表时，是否对已有表结构进行兼容性校验与自动适配。

Boolean

否

true

取值及写入预先创建的表时的行为如下：

  • true（默认值）：校验已有表的列类型兼容性与 NOT NULL 列缺失/多余，并自动补齐上游存在而下游缺失的可空列；出现无法自动适配的情况时作业报错并给出建议执行的 SQL。

  • false：跳过上述校验与自动补列，直接以下游已有表结构为准写出。仅跳过执行初始建表语句时的补列操作，后续的表结构变更会自动执行。

说明

仅实时计算引擎VVR 11.8及以上版本支持。

branch

写入指定的分支。

是

STRING

main

说明

仅实时计算引擎VVR 11.9及以上版本支持。

复用已有 Catalog

自VVR 11.5版本起，您可以在Flink CDC数据摄入作业中直接引用“数据管理”页面中创建的内置Paimon Catalog，减少手写连接属性工作量。

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

目前，数据摄入作业支持自动复用所有 Paimon Catalog 参数，等价于在 YAML 作业中手动配置 catalog.properties.前缀的参数。

如果希望覆盖以上自动复用的参数，可显式写出相应的 YAML 参数，其具备更高的优先级。例如，上面的例子中 Catalog 参数 fs.oss.endpoint总是采纳 YAML 作业中显式写出的选项，而非在 paimon_dlf_catalog中定义的选项。

表结构变更

CDC Yaml Pipeline作业在处理表结构变更时有不同的策略，通过pipeline级别的配置项schema.change.behavior来进行表结构变更同步配置。schema.change.behavior取值有IGNORE、LENIENT、TRY_EVOLVE、EVOLVE 和 EXCEPTION。其中LENIENT和EVOLVE涉及到表结构变更，接下来会说明如何处理不同表结构变更事件（Schema Change Event）。

LENIENT（默认）

LENIENT模式下支持的Schema变更策略详情如下：

  • 添加可空列：会自动在结果表Schema末尾添加对应的列，并自动同步新增列的数据。

  • 删除可空列：不会直接在结果表中删除该列，而是将该列的数据自动填充为NULL值。

  • 添加非空列：会自动在结果表Schema末尾添加对应的列，并自动同步新增列的数据，新增的列会默认设置为可空列，对于添加列发生之前的数据自动设置为NULL值。

  • 重命名列：被看作为添加列和删除列。直接在结果表中末尾添加重命名后的列，并将重命名前的列数据自动填充为NULL值。例如，如果col_a重命名为col_b，则会在结果表末尾添加col_b，并自动将col_a的数据填充为NULL值。

  • 列类型变更：会在结果表中变更对应列的数据类型。

  • 暂不支持同步以下Schema的变更：

    • 主键类型的变更。

    • 从NULLABLE转为NOT NULL变更。

EVOLVE

EVOLVE模式下支持的Schema变更策略详情如下：

  • 添加可空列：支持

  • 删除可空列：不支持。

  • 添加非空列：会在结果表添加可空列。

  • 重命名列：支持，会在结果表将原有列重命名。

  • 列类型变更：会在结果表中变更对应列的数据类型。

  • 暂不支持同步以下Schema的变更：

    • 主键类型的变更。

    • 从NULLABLE转为NOT NULL变更。

开启EVOLVE模式示例请参见开启EVOLVE模式。

说明

在下游 Paimon 表已经存在时，会优先使用已有的表结构进行写入，不会尝试重复建表。在已有的表结构和上游表不一致时：上游存在而下游缺失的可空列会被自动添加到已有表；若出现列类型不兼容、NOT NULL 列缺失或多余等无法自动适配的情况，会出现校验不通过的报错，您可以参考报错中建议执行的 SQL 修改表结构，也可以通过在 sink 中配置 sink.existing-table.schema-check.enabled 为 false 忽略该校验与自动补列，保持以下游表结构为目标进行写出。

使用示例

下面展示几个典型的用户使用场景下的配置示例。

写入 Rest Catalog

Paimon Catalog为rest，写入阿里云数据湖构建的配置示例：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

其中，catalog.properties前缀的参数含义请参见Flink CDC对接Catalog配置参数。

写入 FileSystem Catalog

Paimon Catalog为filesystem，写入阿里云OSS的配置示例：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

其中，catalog.properties前缀的参数含义请参见创建Paimon Filesystem Catalog。

写入分区表

将上游时间戳类型的create_time字段转化为日期类型，作为Paimon表的分区键。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Paimon Pipeline

单表同步

建议为对实时性和稳定性有较高要求的表配置单表同步作业，能够避免多表同步带来的复杂性风险，单表同步的配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整库同步

使用整库同步能力能够降低需要配置和运行的同步作业数，减少运维成本，配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整库同步并替换库表名

在整库同步的基础上，可以增加 route 模块对库名、表名进行统一替换，配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
route:
  # 将 MySQL test_db 数据库中的所有表同步到 Paimon test_db2 库中，并保留表名不变
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Paimon Pipeline

分库分表合并

在您有将多张分表的数据写入到一张下游表需求的场景下，可以使用如下的配置模板：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
route:
  # MySQL test_db 中的所有分表将被合并到一张 Paimon test_db.user 表中。
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Paimon Pipeline

重启新增存量表

如果想要新增同步存量表，设置scan.newly-added-table.enabled = true后重启作业。

重要

如果作业先设置scan.binlog.newly-added-table.enabled为true捕获新增表，不可以再通过scan.newly-added-table.enabled = true重启捕获存量表，否则会有数据重复发送。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # 在作业重启时，检查 tables 参数捕获的新表并执行快照。
  # 注意需要搭配 scan.startup.mode: initial 参数使用。
  scan.newly-added-table.enabled: true
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整库同步时排除部分表

在整库同步的场景下，如果有部分表（测试表或者某些会导致作业持续 Failover 的表）需要排除，配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # 此正则命中的表都不会被同步。
  tables.exclude: test_db.table1
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

整库同步到文件格式为Lance的Paimon表

Lance是一种专为向量和多模态数据设计的数据存储格式，您可以通过设置file.format的建表参数以写到文件格式为Lance的Paimon表，配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  # 指定写出文件格式为 lance 
  table.properties.file.format: lance
pipeline:
  name: MySQL to Paimon Pipeline

开启EVOLVE模式

开启 EVOLVE 模式能够保持下游表与上游表表结构严格一致，包括执行与上游相同的删除字段、删除表等操作。在此模式下，若下游无法支持应用所有的表结构变更事件，可能导致作业Failover且无法自愈。作业配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline
  schema.change.behavior: evolve

常见问题

上一篇：Paimon SQL连接器下一篇：视频流（RTSP/RTMP/HLS）（公测中）