在 Flink 控制台，bytehouse-ce 连接器支持做结果表，可以通过 Flink 任务将数据写入到 ByteHouse 目标表。

背景信息

ByteHouse 是一款云原生数据仓库，是火山引擎基于开源 ClickHouse 进行深度优化和改造的版本，提供海量数据上更强的查询服务和数据写入性能。
ByteHouse 企业版（CE）基于火山内部的丰富场景，以及 ClickHouse 开源版的痛点进行了深度定制，包括多场景表引擎、扩展数据类型、多级存储等功能。如需了解 ByteHouse 企业版更多信息，请参见ByteHouse 企业版简介。

使用限制

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

DDL 定义

CREATE TABLE bh_ce_sink (
  f0 VARCHAR,
  f1 VARCHAR,
  f2 VARCHAR
) WITH (
  'connector' = 'bytehouse-ce',
  'clickhouse.cluster' = 'bytehouse_cluster_***',
  'clickhouse.shard-discovery.service.host' = '7249621***.bytehouse-ce.ivolces.com',
  'username' = 'user_a',
  'password' = 'pa***45',
  'database' = 'default',
  'table-name' = 'doc_test',
  'sink.buffer-flush.interval' = '10 second',
  'sink.buffer-flush.max-rows' = '5000',
  'clickhouse.shard-discovery.address-mapping' = '192.18.*.*|8123:192.168.*.*|8123'
);

数据类型映射

如下表格是 Flink SQL 数据类型和 ByteHouse 数据类型的映射关系。如果需要了解

WITH 参数

示例代码

CREATE TABLE random_source (
   f0 VARCHAR,
   f1 VARCHAR,
   f2 VARCHAR
 ) WITH (
   'connector' = 'datagen',
   'rows-per-second' = '1'
 );

CREATE TABLE bh_ce_sink (
  f0 VARCHAR,
  f1 VARCHAR,
  f2 VARCHAR
) WITH (
  'connector' = 'bytehouse-ce',
  'clickhouse.cluster' = 'bytehouse_cluster_***',
  'clickhouse.shard-discovery.service.host' = '7249621***.bytehouse-ce.ivolces.com',
  'username' = 'user_a',
  'password' = 'pa***45',
  'database' = 'default',
  'table-name' = 'doc_test',
  'sink.buffer-flush.interval' = '10 second',
  'sink.buffer-flush.max-rows' = '5000',
  'clickhouse.shard-discovery.address-mapping' = '192.18.*.*|8123:192.168.*.*|8123'
);

INSERT INTO bh_ce_sink 
SELECT f0, f1, f2 FROM random_source;

常见问题

导入 ByteHouse 后出现未来时间

问题描述：

1. 上游 Kafka 数据源中，有字段为字符串格式时间戳例如 {"t": "2024-10-17 10:00:00"}
2. Flink 的 DDL 语句中定义 t字段为 TIMESTAMP 类型。
3. ByteHouse 中下游数据类型为 DateTime 或 DateTime64 两种类型。
4. 写入 ByteHouse 中时间戳查询结果为 2024-10-17 18:00:00

问题原因：源端（例如 Kafka 数据源）时间戳数据以不带时区的格式输出时（如 “2024-10-17 10:00:00” ），默认按 UTC 解析为 Epoch 时间戳；如果源端业务的时区并非 UTC，那么按上述方式输出时间戳值则会间接地引入 UTC 与源端业务时区之间的小时数偏差。

解决方案：

1. 推荐方案一：Kafka 数据源中时间戳建议使用 UTC 时间例如 "2024-10-17T10:00:00Z"，同时 Flink 建表语句中定义字段类型为 TIMESTAMP_LTZ。
2. 推荐方案二：在 Flink ByteHouse 建表语句 WITH OPTIONS 中增加 'timestmap-offset' = '-8h'，这样子在单个 ByteHouse Sink 中生效。
3. 推荐方案三：在 Flink 任务中统一设置变量，这个和方案二效果类似，但是对 Flink 任务中所有的 ByteHouse Sink 都生效containerized.taskmanager.env.FLINK_WRITE_TO_BYTEHOUSE_TIMESTAMP_OFFSET: -8h