Kafka 到 StarRocks

选择对端数据库：

StarRocks

数据链路

基本功能

功能	说明
Incremental Data Sync	Allow subscribing to messages from the source Topic and transforming them into DML operations, namely, INSERT, UPDATE, DELETE.
Subscription Modification	Add, delete, or modify the subscribed topics. For more information, see Modify Subscription.
Position Resetting	Reset positions by timestamp to consume the data in a past period again.

高级功能

功能	说明
消息格式	支持以下消息格式，文档：消息格式说明 CloudCanal内置格式 AlibabaCanal兼容格式
全量前清空目标数据	运行全量任务前清除老数据，包括重跑任务、定时全量迁移都会触发此能力
重建目标表	运行全量任务前重建目标表，包括重跑任务、定时全量迁移都会触发此能力
Stream Load 数据写入	采用 Stream Load 到 StarRocks Be 写入数据, 默认攒批写入，可动态调节刷出数据节奏和批次大小
0 值时间处理	支持将 0 值时间设置成不同类型的值，防止写入对端报错

限制和注意点

限制项	说明
目标端需要提前创建表	仅支持消息自动创建 Topic
原始消息格式	仅支持 Kafka 到 Kafka，且两端的消息格式都需要选择 原始消息格式
对端表类型	仅支持 主键模型(Primary Key)
源端表类型	不支持 无主键表 迁移同步
DDL 同步报错	同一张表连续几个 DDL 将报错（因 StarRocks 对端是异步 DDL) 修改字段约束或者部分类型的 DDL 报错 如遇到 DDL 报错，可在对端变更好表结构，然后通过设置任务参数跳过，文档：跳过 DDL 异常
增量写入冲突策略	Stream Load 写入以主键进行整行替换

使用示例

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

链路FAQ

• 如何打开 Kafka 源端心跳?

• Doris / SelectDB 目标端写入报错

• StarRocks 对端结构迁移如何设置分区?

---

源端数据源

前置条件

条件	说明
网络准备	迁移同步节点（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: 通用参数配置请参考 通用参数及功能

---

目标端数据源

前置条件

条件	说明
账号权限	具备 SELECT, DDL 权限（可选）
网络准备	迁移同步节点（sidecar）可连接 StarRocks FE QueryPort 和 FE/BE HttpPort

任务参数

参数名称	说明
host	MySQL 协议交互链接，对应 StarRocks FE QueryPort
httpHost	StarRocks stream load 链接，对应 StarRocks FE/BE HttpPort
totalDataInMemMb	攒批写入，内存中最大数据容量，超过此容量或超过 asyncFlushIntervalSec 则刷出数据到写入队列
asyncFlushIntervalSec	攒批写入，等待刷出的间隔时间，超过此时间或超过 totalDataInMemMb 则刷出数据到写入队列
flushBatchMb	单表最大攒批容量，超过此容量则刷出数据到写入队列
realFlushPauseSec	使用 stream load 刷出数据到 StarRocks 的等待时间，0 则不等待
soTimeoutSec	在 QueryPort 执行操作时 tcp 超时链接 (so_timeout)
httpSoTimeoutSec	在 HttpPort 执行操作时 tcp 超时链接 (so_timeout)
enableTimeZoneProcess	是否对时间字段进行时区转换
timezone	目标端 StarRocks 时区，例如 +08:00 Asia/Shanghai America/New_York
maxInSizePerQuery	校验任务中，对端单次查询的最大 IN 条件值数量，大于该值会自动拆分多次查询

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