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

文档中心

立即注册

导航

Bytehouse CDW

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

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

使用限制

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

连接器的功能

数据实时同步（包括整库同步、分库分表同步）

如何创建 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
  timestamp-offset: -8h

pipeline:
  name: MySQL to Bytehouse CDW Pipeline
  parallelism: 16

连接器配置项

参数	是否必选	默认值	数据类型	描述
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 偏移，来校正数据。

使用说明

目前不支持自动建表，原因是 Bytehouse CDW 建表参数较多，自动建的表不可用，所以推荐手动建表，并对建表参数进行调优后，启动 Flink CDC 3.0 任务导入数据。
对于 Flink 类型为 TIMESTAMP 和 TIMESTAMP WITH LOCAL TIME ZONE 的字段数据，目前在进入 Bytehouse CDW Sink 前，都会转为本地的时间戳，由于 Bytehouse CDW Sink 是按照 UTC 来解析该时间戳，因此需要加 -8h 偏移，来校正数据。

数据类型映射

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