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

背景信息

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

使用限制

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

DDL 定义

CREATE TABLE bh_cdw (
    f0 VARCHAR, 
    f1 VARCHAR, 
    f2 VARCHAR) WITH (
    'connector' = 'bytehouse-cdw',
    'database' = 'doc_db',
    'table-name' = 'doc_table_2',
    'username' = 'user-a',
    'password' = 'qa***6',
    -- 指定 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.virtual-warehouse' = 'test',
    'jdbc.enable-gateway-connection' = 'true',
    'bytehouse.gateway.account' = '210***34',
    'bytehouse.gateway.access-key-id' = '<your-access-key>',
    'bytehouse.gateway.secret-key' = '<your-secret-key>',
    'sink.buffer-flush.interval' = '5 second',
    'sink.buffer-flush.max-rows' = '2000'
);

WITH 参数

数据类型映射

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

示例代码

SQL 代码示例

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


CREATE TABLE bh_cdw (
    f0 VARCHAR, 
    f1 VARCHAR, 
    f2 VARCHAR) WITH (
    'connector' = 'bytehouse-cdw',
    'database' = 'doc_db',
    'table-name' = 'doc_table_2',
    'username' = 'user-a',
    'password' = 'qa***6',
    -- 指定 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.virtual-warehouse' = 'test',
    'jdbc.enable-gateway-connection' = 'true',
    'bytehouse.gateway.account' = '210***34',
    'bytehouse.gateway.access-key-id' = '<your-access-key>',
    'bytehouse.gateway.secret-key' = '<your-secret-key>',
    'sink.buffer-flush.interval' = '5 second',
    'sink.buffer-flush.max-rows' = '2000'
);


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

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