Bytehouse CDW--流式计算 Flink版-火山引擎

文档中心

立即注册

导航

Bytehouse CDW

最近更新时间：2025.09.18 14:43:30首次发布时间：2024.10.23 20:28:33

Bytehouse CDW Pipeline 连接器可以用作 Pipeline 的 Data Sink，将数据写入 Bytehouse CDW。本文档介绍如何设置 Bytehouse CDW Pipeline 连接器。

使用限制

Bytehouse CDW Pipeline 连接器暂时仅支持在 Flink 1.16-volcano 引擎版本中使用。

连接器的功能

数据实时同步（包括整库同步、分库分表同步）
自动建表，并支持在建表时设置 table settings 参数

如何创建 Pipeline

从 MySQL 读取数据同步到 Bytehouse CDW 的 Pipeline 可以定义如下：

################################################################################
# Description: Sync MySQL all tables to BH CDW
################################################################################
source:
   type: mysql
   name: MySQL Source
   hostname: vedbm-xxx.pri.mysql.vedb.ivolces.com
   port: 3306
   username: your-username
   password: your-password
   tables: app_db.\.*
   server-id: 5401-5420

sink:
  type: bytehouse-cdw
  name: Bytehouse CDW Sink
  region: VOLCANO_PRIVATE
  host: tenant-xxxx-cn-beijing.bytehouse.ivolces.com
  port: 19000
  virtual-warehouse: test
  api-token: xxx:xxx
  sink.buffer-flush.interval: 5s
  sink.buffer-flush.max-rows: 10000

pipeline:
  name: MySQL to Bytehouse CDW Pipeline
  parallelism: 4 # 注意作业并发度以该参数为准，Flink 平台参数配置页面的并发度不生效

连接器配置项

参数	是否必选	默认值	数据类型	描述
type	required	(none)	String	指定要使用的连接器, 这里需要设置成 `bytehouse-cdw`.
name	optional	(none)	String	Sink 的名称.
region	optional	`VOLCANO_PRIVATE`	String	指定 ByteHouse Gateway 的地域。 `VOLCANO_PRIVATE`：火山引擎私有网络，此时需要 ByteHouse CDW和 Flink 处于相同 VPC。以及填写私有域名地址和端口。
host	required	(none)	String	ByteHouse 网关的私有主机。前提是将 region 设置为 `VOLCANO_PRIVATE`。
port	optional	19000	String	ByteHouse 网关的私有端口。前提是将 region 设置为 `VOLCANO_PRIVATE`。
virtual-warehouse	required	(none)	String	用于指定计算组的名称。
api-token	required	(none)	String	连接器帐户的 API Token。
sink.mode	optional	upsert	String	选择要接收的数据记录。支持的值包括： insert：仅接受 INSERT 记录。 upsert：接受 INSERT、UPDATE_AFTER 和 DELETE 记录。 upsert-no-delete：接受 INSERT 和 UPDATE_AFTER 记录。 upsert-all：接受所有类型的记录。 upsert-all-no-delete：接受除 DELETE 之外的所有类型的记录。 upsert-with-delete-to-null：接受 INSERT 和 UPDATE_AFTER 记录。对于 DELETE，它将除 UPSERT 键（包括分区键和唯一键）之外的所有字段设置为 NULL。 delete-only：仅接受 DELETE 记录。
sink.buffer-flush.interval	optional	1s	Duration	刷新时间间隔，最小值为 200 ms。
sink.buffer-flush.max-rows	optional	50000	Integer	缓冲记录大小，最小值为 100。
sink.max-retries	optional	15	Integer	最大写入重试次数，设置为 -1 表示无限次重试。
jdbc.max-retry-backoff	optional	32s	Duration	最大指数回退重试间隔，会从 1s 开始指数递增。默认为 32s，表示从 1s、2s、4s ...，一直递增到 32s 后，保持不变。
timestamp-offset	optional	(none)	Duration	TIMESTAMP 类型的附加时间偏移。注意：对于 Flink 类型为 TIMESTAMP 和 TIMESTAMP WITH LOCAL TIME ZONE 的字段数据，目前在进入 Bytehouse CDW Sink 前，都会转为本地的时间戳，由于 Bytehouse CDW Sink 是按照 UTC 来解析该时间戳，因此需要加 -8h 偏移，来校正数据。

自动建表

自动建表默认开启，会自动通过上游 DDL，完成在 Bytehouse CDW 的建表。
当前支持的功能：

支持所有常见数据类型，包括基本类型和复杂类型
支持主键表和非主键表
支持在建表时设置 table settings 参数

注意：对于大表、对查询性能要求严格的表，推荐采用手动建表，并对 Bytehouse 建表参数进行调优，以满足查询和写入性能要求。这是由于自动建表只设置了 unique key 属性，其他属性没有设置，包括 order by、cluster by 等，如果性能严格较高，则推荐手动建表。
此外，支持透传 settings，在建表时提供更多的参数配置，使用方式如下：
需要在 sink option 里添加如下参数：

sink:
  type: bytehouse-cdw
  name: Bytehouse CDW Sink
  ...
  table.create.settings: enable_bucket_level_unique_keys=1,max_staged_part_rows_per_task=150000000000,storage_dialect_type='MYSQL',cloud_enable_staging_area=true

value 如果有多个配置，则通过逗号分隔。
另外，也可以通过 Transform 算子指定 table options：

transform:
  - source-table: app_db.orders_full
    projection: "*"
    table-options: storage_policy='cnch_default_hdfs',index_granularity=1024
    description: auto creating table options example

table-options 的格式是 k1=v1,k2=v2，注意如果 value 是 string，需要用单引号括起来，如果是数字或者 bool 值，则不需要。
Bytehouse CDW 所有 table settings 的文档参见：https://www.volcengine.com/docs/6517/1555178

数据类型映射

CDC type	Bytehouse CDW type	NOTE
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
DATE	DATE
TIME	TIME	Bytehouse CDW Connector 目前不支持 TIME 类型写入，预计下个版本支持。
TIMESTAMP	DATETIME
TIMESTAMP_LTZ	DATETIME
CHAR(n)	STRING
VARCHAR(n)	STRING
STRING	STRING
ARRAY	ARRAY	Bytehouse CDW Connector 目前不支持 ARRAY 类型写入，预计下个版本支持。