ByteHouse CE Pipeline Sink Connector 用于将 Flink CDC Pipeline 产生的全量数据、增量数据变更和表结构变更写入 ByteHouse CE。本 Connector 支持自动创建目标表，并支持在建表时设置 Table Settings。

• 自动建表：目标表不存在时，根据上游表结构创建 ByteHouse CE 本地表和 Distributed 表。
• 表结构变更同步：支持加列、删列、重命名列、修改列类型的 Schema 演进。
• 多种写入模式：可通过 sink.mode 配置项来控制新增、更新和删除数据的处理方式。
• 批量写入与重试：​可配置批量 flush 间隔与行数、最大重试次数、重试退避和写入限速。
• 主键 Upsert：​有主键表自动创建为 HaUniqueMergeTree，支持按主键更新覆盖和删除。

• 已开通火山引擎流式计算 Flink 版，并创建可用的 Flink 资源池。
• 已创建 ByteHouse CE 集群，并准备好连接信息：host（API Cluster Discovery Service 地址）、集群名（cluster-name）、火山账号 ID（account-id）、用户名和密码。
• 已确保 Flink 资源池能够访问 ByteHouse CE Cluster Discovery Service 。
• 使用自动建表功能时，连接账号需具备创建数据库、本地表和 Distributed 表的权限；目标表已存在时，账号需具备对应表的读写权限。
• 已配置上游 Source（如 MySQL）；upsert/delete 同步场景下，源表需定义稳定、唯一且非空的主键。

以下示例将 MySQL 的 app_db 全库表同步到 ByteHouse CE：

sources:
- source:
    type: mysql
    name: MySQL Source
    hostname: mysqlxxxxxxxx.rds.ivolces.com
    port: 3306
    username: your-username
    password: your-password
    tables: app_db.\.*
    server-id: 5100-5104

sink:
  type: bytehouse-ce
  name: ByteHouse CE Sink
  host: xxxxxxxxxx.bytehouse-ce.ivolces.com
  port: 80
  cluster-name: your-ce-cluster
  account-id: your-account-id
  username: your-ce-username
  password: your-password

pipeline:
  name: mysql-to-bytehouse-ce
  schema.change.behavior: evolve
  parallelism: 2

连接与认证参数 #

写入与重试参数 #

Sink Mode 选择 #

自动建表参数 #

• 本地表名称为目标表名加 _local，Distributed 表沿用目标表名。
• Pipeline partition key 映射为本地表的 PARTITION BY。
• Pipeline 表注释与列注释会传入建表 DDL。
• 目标表已存在时，Connector 跳过创建；非 insert 模式会校验其是否具备 unique key。
• 目标表不存在且 table.create.enabled: false 时，作业会快速失败。

主要类型映射如下：

完整示例 #

以下示例将 MySQL 的 app_db 全库表同步到 ByteHouse CE。运行前请替换 MySQL 连接信息和 ByteHouse CE 连接信息。

sources:
  - source:
      type: mysql
      name: MySQL Source
      hostname: <mysql-host>
      port: 3306
      username: <mysql-username>
      password: <mysql-password>
      tables: app_db.\.*
      server-id: 5100-5104

sink:
  type: bytehouse-ce
  name: ByteHouse CE Sink
  host: <bytehouse-ce-discovery-host>
  port: 80
  cluster-name: <bytehouse-ce-cluster-name>
  account-id: <volcengine-account-id>
  username: <bytehouse-ce-username>
  password: <bytehouse-ce-password>

  # 标准 CDC 场景：写入新增、更新和删除记录。
  sink.mode: upsert

  # 目标表不存在时，根据上游 Schema 自动创建。
  table.create.enabled: true
  table.create.settings: index_granularity=8192,enable_unique_partial_update=1

  # 攒批与重试参数，可根据吞吐和延迟要求调整。
  sink.buffer-flush.interval: 1s
  sink.buffer-flush.max-rows: 20000
  sink.max-retries: 15
  jdbc.max-retry-backoff: 32s

pipeline:
  name: mysql-to-bytehouse-ce
  schema.change.behavior: evolve
  parallelism: 2

预期结果：

• 全量阶段：上游 app_db 已有表的数据被批量写入 ByteHouse CE。有主键表自动创建为 HaUniqueMergeTree 本地表并配套 Distributed 表；无主键表自动创建为 HaMergeTree本地表并配套 Distributed 表。
• 增量阶段：上游 INSERT、UPDATE 转换为 RowKind.INSERT 并按主键 Upsert 写入；DELETE 转换为 RowKind.DELETE 并删除对应记录（默认 upsert 模式）。
• 最终状态：在 sink.mode: upsert 且目标表具备主键时，同一主键的多次变更会被覆盖更新，目标表最终反映上游的当前状态。

同步无主键表 #

无主键表不能使用 Upsert 语义。仅追加场景需要将写入模式改为 insert：

sink:
  type: bytehouse-ce
  # 省略连接参数
  sink.mode: insert
  table.create.enabled: true

忽略删除事件 #

如果只需同步新增和更新记录，可使用 upsert-no-delete：

sink:
  type: bytehouse-ce
  # 省略连接参数
  sink.mode: upsert-no-delete

预先创建目标表 #

大表或查询性能敏感表建议先在 ByteHouse CE 中完成表结构、分区、排序键、TTL 和索引设计，再关闭自动建表：

sink:
  type: bytehouse-ce
  # 省略连接参数
  table.create.enabled: false

数据事件 #

Connector 将 Pipeline 事件转换为 RowKind 后批量写入 ByteHouse CE，转换规则如下：

• sink.buffer-flush.interval 与 sink.buffer-flush.max-rows 控制批量 flush 的最大间隔和缓存行数。
• sink.max-retries 控制写入 flush 与 CE metadata client 操作的最大重试次数，jdbc.max-retry-backoff 控制重试延迟上限。
• max-rows-per-second 可限制每个 Sink subtask 的写入速率；clickhouse.query.settings 可向写入查询追加 ClickHouse query settings。

自动建表 #

设置建表参数：

sink:
  type: bytehouse-ce
  # 省略连接参数
  table.create.enabled: true
  table.create.settings: index_granularity=8192,enable_unique_key_partial_update=1
  table.create.distributed-settings: max_parallel_replicas=2

table.create.settings 应用于本地表，table.create.distributed-settings 应用于 Distributed 表。多个配置以逗号分隔；字符串值需要按 ByteHouse DDL 语法加引号。
也可以通过 transform 为单张表指定 schema options，这些选项会追加到本地表 SETTINGS：

transform:
  - source-table: app_db.orders
    projection: "*"
    table-options: ttl_only_drop_parts=1,index_granularity=8192
    description: ByteHouse CE auto-create options

Schema 事件 #

将 Pipeline 配置为 schema.change.behavior: evolve，Sink 接收并处理表结构变更事件(Schema Change Event) 的行为如下：

Checkpoint 与故障恢复 #

• 写入失败时 Connector 按 sink.max-retries 自动重试；超过重试次数后作业失败，可结合 Flink restart strategy 控制作业级恢复。
• 故障恢复可能导致部分记录被重放，写入链路采用 At-Least-Once 语义。
• 使用默认 upsert 模式并为目标表配置主键时，重复 INSERT 会按主键覆盖，目标表最终状态保持一致；insert 追加模式下重复写入可能产生多行。

验证方法 #

首次使用 ByteHouse CE Pipeline Sink 时，建议使用测试库表和测试集群完成全量同步、增量同步、Schema 演进和故障恢复验证，再应用于生产环境。建议按照以下步骤进行验证：

1. 检查 host、port、cluster-name、account-id、用户名和密码是否正确，并确认 Flink 资源池到 ByteHouse CE discovery service 网络互通。
2. 确认作业引擎版本为 Flink 1.16-volcano 及以上。
3. 提交 Pipeline，确认作业进入运行状态，目标数据库和表已自动创建。
4. 检查全量数据，确认上游已有数据已写入目标表。
5. 在上游依次执行插入、更新和删除操作，确认下游收到对应变更。
6. 在上游执行加列或修改列类型，确认 Schema 演进生效。
7. 等待至少一次 Checkpoint 成功后执行受控重启，确认作业恢复后目标表最终状态一致。

常见问题 #

默认 upsert 自动建表报“requires a source primary key” #

上游 Schema 没有主键，导致 upsert 模式无法自动建表。可通过以下方式解决：

• 在源表声明中显式指定主键；
• 将写入模式改为 sink.mode: insert（需显式设置，系统不会自动切换）；
• 提前在 ByteHouse CE 中手动创建带有 unique key 的目标表，并关闭自动建表（table.create.enabled: false），让任务直接写入已有表。

目标表已存在，但报“has no unique key” #

除 insert 外的模式都要求目标表具有 unique key。检查是否误连到普通 MergeTree 表，或改为预建 HaUniqueMergeTree/unique-key 表。

ZooKeeper path 校验失败 #

table.create.zookeeper-path 必须同时包含 ${database} 与 ${table}。保留 {shard} 宏可让 CE 为各 shard 解析路径。

主键列名导致自动建表失败 #

自动生成 cityHash64(...) 时 SDK 不为 sharding 表达式里的列名加引号，因此主键列名只能使用字母、数字和下划线，且不能以数字开头。可通过 Transform 规范字段名，或手动建表并关闭自动建表。

设置 sink.max-retries: -1 后校验失败 #

CE Connector 不支持以 -1 表示无限重试。请配置大于等于 0 的有限次数，并结合 Flink restart strategy 控制作业级恢复。

时间字段相差 8 小时 #

先核对 Source 的时区配置、Pipeline 的 local-time-zone 和 ByteHouse 字段类型。仅在确认 SDK 写入需要固定修正时配置 timestamp-offset，不要无条件套用 -8h。

数据没有删除 #

检查 sink.mode 是否为 upsert-no-delete、upsert-all-no-delete 或 insert；这些模式不会把 DELETE 应用为目标表删除。需要同步删除时，将 sink.mode 改为默认 upsert，并确认目标表具备 unique key。

无法发现 shard 或连接失败 #

检查 host、port、cluster-name、account-id，以及 Flink 所在网络到 discovery service 的连通性。认证失败时再核对用户名、密码和账号权限。

• Flink CDC Pipeline 连接器列表：了解火山引擎 Flink CDC Pipeline Connector 的文档分类和支持范围。
• MySQL Pipeline Connector：了解示例中 MySQL Source 的配置参数和数据读取方式。
• Apache Flink CDC Data Pipeline：了解 Flink CDC Pipeline 中 Source、Sink、Transform、Route 和 Pipeline 的关系。
• ByteHouse CE HaUniqueMergeTree：了解 ByteHouse CE HaUniqueMergeTree 表引擎相关概念。