Kafka 到 Paimon

CloudCanal 支持从 Kafka 到 Paimon 的数据迁移、同步、校验和链路能力。

选择对端数据库：

Paimon

数据链路

基本功能

功能	说明
增量实时同步	支持订阅源端 Topic 的消息，并转换为 INSERT、UPDATE、DELETE DML
修改订阅	新增、删除、修改订阅 Topic，文档：修改订阅
重置位点	按 时间戳 回溯位点，重新消费过去一段时间的数据

高级功能

功能	说明
消息格式	支持以下消息格式，文档：消息格式说明 CloudCanal内置格式 AlibabaCanal兼容格式
写入冲突策略	源端有主键表进行覆盖写入，源端无主键表进行追加写入
自定义表属性	包括 format-version 等属性设置
设置数据分区	创建任务时可按表级粒度指定分区字段名（多个以逗号分隔，如：col1,col2,col3），结构迁移过程中将自动添加对应分区

限制和注意点

限制项	说明
目标端需要提前创建表	仅支持消息自动创建 Topic
原始消息格式	仅支持 Kafka 到 Kafka，且两端的消息格式都需要选择 原始消息格式

使用示例

标题	详情
跨互联网数据互通 (Kafka)	文档：跨互联网数据互通 (Kafka)
Kafka 数据中转校验	文档：Kafka 数据中转校验

链路FAQ

• 如何打开 Kafka 源端心跳?

---

源端数据源

前置条件

条件	说明
网络准备	迁移同步节点（sidecar）可连接 Kafka 各节点

任务参数

参数名称	说明
schemaFormat	MQ 消息格式，文档：消息格式说明
consumerGroupId	Kafka 消费组 Id
consumeParallel	消费 Kafka 的并行度
sessonTimeoutMs	Kafka Session 超时时间（毫秒）
maxPollRecords	Kafka 一次最大拉取消息数量
dbHeartbeatIntervalSec	配置对源端数据库发起心跳操作的间隔时长
dbHeartbeatToleranceStep	配置对源端数据库心跳操作可容忍的位点差值
customClientProps	自定义传入到 Kafka Client 参数，JSON 格式，key为参数名，value为参数值。此配置项以最高优先级生效。例如：AWS IAM 访问控制

Tips: 通用参数配置请参考 通用参数及功能

---

目标端数据源

前置条件

条件	说明
网络准备	迁移同步节点（sidecar）可连接 Catalog 和 Warehouse
S3 数据源配置模版	网络地址（CatalogUri）: glue.ap-southeast-1.amazonaws.com httpsEnabled: true metastoreType: filesystem warehouse : s3://warehouse/test catalogProps: { "s3.access-key": "", "s3.secret-key": "", "s3.path-style-access": "true" }

任务参数

参数名称	说明
writeParallel	对端写入并发数，对迁移或者同步性能影响大
writeProps	写入配置参数（Json 格式）
compactTaskSize	压缩任务线程数
paimonIOManagerPaths	I/O 管理器路径
defaultDynamicBucket	默认动态 Bucket 数
totalDataInMemMb	攒批写入，内存中最大数据容量，超过此容量或超过 asyncFlushIntervalSec 则刷出数据到写入队列
asyncFlushIntervalSec	攒批写入，等待刷出的间隔时间，超过此时间或超过 totalDataInMemMb 则刷出数据到写入队列
flushBatchMb	单表最大攒批容量，超过此容量则刷出数据到写入队列
realFlushPauseSec	刷出数据到 Paimon 的等待时间，0 则不等待
catalogUri	Catalog 地址
httpsEnabled	Catalog URI 使用 HTTPS
catalogWarehouse	指定 Paimon 仓库的根目录 URI
catalogMetastoreType	启用文件系统模式
catalogProps	Catalog 配置参数
enableTimeZoneProcess	是否对时间字段进行时区转换
timezone	目标端时区，例如 +08:00 Asia/Shanghai America/New_York

Tips: 通用参数配置请参考 通用参数及功能