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

背景信息

ByteHouse 是一款云原生数据仓库，云数仓版（CDW）是一个支持实时导入和离线导入的自助数据分析平台，能够对海量数据进行高效分析。
如需了解 ByteHouse 云数仓版更多信息，请参见 ByteHouse 云数仓版简介。

使用限制

• ByteHouse CDW 连接器暂时仅支持在 Flink 1.16-volcano 引擎版本中使用。
• 如果使用火山引擎私有网络，此时需要 ByteHouse CDW 和 Flink 处于相同 VPC，具体设置方法可以参考 ByteHouse CDW 网络设置文档。
• 如果您需要使用更新版本的 ByteHouse Flink Connector ，可以在 ByteHouse CDW Driver 进行下载。

DDL 定义

CREATE TABLE `bh_de_sink` (  
  `id` STRING NOT NULL,
  `event_time` STRING,
  `content` ARRAY<DECIMAL(20, 0)>
)
PARTITIONED BY (`id`)
WITH (
  'connector' = 'bytehouse-cdw',
  'jdbc.enable-gateway-connection' = 'true',
  -- 指定 ByteHouse Gateway 的地域。
  -- 示例VOLCANO_PRIVATE为火山引擎私有网络，此时需要ByteHouse CDW和Flink处于相同VPC。
  'bytehouse.gateway.region' = 'VOLCANO_PRIVATE',
  'bytehouse.gateway.host' = 'tenant-xxxx.bytehouse.ivolces.com',
  'bytehouse.gateway.port' = '19000',
  'bytehouse.gateway.api-token' = '<< your API token >>',
  'bytehouse.gateway.virtual-warehouse' = 'default_vw',
  'database' = 'demo_database',
  'table-name' = 'demo_table'
);

WITH 参数

数据类型

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

示例代码

SQL 代码示例

CREATE TABLE `bh_de_source` (  -- source table
  `id` BIGINT NOT NULL,
  `time` TIMESTAMP(0),
  `content` ARRAY<DECIMAL(20, 0)>
)
PARTITIONED BY (`id`)
WITH (
  'connector' = 'datagen'
);

CREATE TABLE `bh_de_sink` (  -- sink table
  `id` STRING NOT NULL,
  `event_time` STRING,
  `content` ARRAY<DECIMAL(20, 0)>
)
PARTITIONED BY (`id`)
WITH (
  'connector' = 'bytehouse-cdw',
  'jdbc.enable-gateway-connection' = 'true',
  'bytehouse.gateway.region' = 'VOLCANO_PRIVATE',
  'bytehouse.gateway.host' = 'tenant-xxxx.bytehouse.ivolces.com',
  'bytehouse.gateway.port' = '19000',
  'bytehouse.gateway.api-token' = '<< your API token >>',
  'bytehouse.gateway.virtual-warehouse' = 'default_vw',
  'database' = 'demo_database',
  'table-name' = 'demo_table'
);

INSERT INTO `bh_de_sink` (  -- data loading
  `id`,
  `event_time`,
  `content`
)
SELECT
  IFNULL(CAST(`id` AS STRING), '43'),  -- type casting with default value
  CAST(`time` AS STRING),  -- type casting
  IFNULL(`content`, 'hello')  -- default value
FROM `bh_de_source`

Java DataStream 代码示例

Flink 连接器的 DataStream API 源数据类型为 RowData。通过 DataStream API 的使用主要流程是通过 CnchSinkFunctionBuilder 获取 CnchSinkFunction 实例，下面是一个演示基本用法的示例。详细参数配置可以参考 ByteHouse 官方文档。

StreamExecutionEnvironment env = ...;

// Adding a source to the data stream
DataStream<RowData> dataStream =
    env.addSource(...).returns(TypeInformation.of(RowData.class));

// List of columns representing the table schema
List<Column> columns =
    Arrays.asList(
        Column.physical("year", DataTypes.INT()),
        Column.physical("npc", DataTypes.STRING()),
        Column.physical("offence", DataTypes.STRING()),
        Column.physical("case_no", DataTypes.INT()));

try (@SuppressWarnings("unchecked")
    CnchSinkFunction<RowData, ?> cnchSink =
        new CnchSinkFunctionBuilder.Insert(database, tableName)
            .withSchema(columns)
            .withGatewayConnection("VOLCANO_PRIVATE", "your_host", "your_port")
            .withGatewayApiToken("<< your API token >>")
            .withGatewayVirtualWarehouse("default_vw")
            .withFlushInterval(Duration.ofSeconds(1))
            .build()) {

    // Add the sink to the data stream
    dataStream.addSink(cnchSink);

    // Trigger the execution
    env.execute();
}

常见问题

导入 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。