在 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` ( -- sink table
 `id` STRING NOT NULL,
 `event_time` STRING,
 `content` ARRAY<DECIMAL(20, 0)>,
 PRIMARY KEY (`id`) NOT ENFORCED -- 主键字段对应 BH 唯一键表的 Unique Key，在 BH 入库的时候会进行去重
) WITH (
 'connector' = 'bytehouse-cdw',
 'jdbc.enable-gateway-connection' = 'true',
 'bytehouse.gateway.region' = 'VOLCANO_PRIVATE',
 'bytehouse.gateway.host' = 'tenant-xxxx-cn-beijing.bytehouse.ivolces.com',
 'bytehouse.gateway.port' = '19000',
 'bytehouse.gateway.api-token' = '<API_KEY>',
 'bytehouse.gateway.virtual-warehouse' = '<VIRTUAL_WAREHOUSE>',
 'database' = '<BYTEHOUSE_DB>',
 'table-name' = '<BYTEHOUSE_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)>
) WITH (
 'connector' = 'datagen'
);

CREATE TABLE `bh_de_sink` ( -- sink table
 `id` STRING NOT NULL,
 `event_time` STRING,
 `content` ARRAY<DECIMAL(20, 0)>,
 PRIMARY KEY (`id`) NOT ENFORCED -- 主键字段对应 BH 唯一键表的 Unique Key
) WITH (
 'connector' = 'bytehouse-cdw',
 'jdbc.enable-gateway-connection' = 'true',
 'bytehouse.gateway.region' = 'VOLCANO_PRIVATE',
 'bytehouse.gateway.host' = 'tenant-xxxx-cn-beijing.bytehouse.ivolces.com',
 'bytehouse.gateway.port' = '19000',
 'bytehouse.gateway.api-token' = '<API_KEY>',
 'bytehouse.gateway.virtual-warehouse' = '<VIRTUAL_WAREHOUSE>',
 'database' = '<BYTEHOUSE_DB>',
 'table-name' = '<BYTEHOUSE_TABLE>'
);

INSERT INTO `bh_de_sink` ( -- data loading
 `id`,
 `event_time`,
 `content`
)
SELECT
 CAST(`id` AS STRING), 
 CAST(`time` AS STRING), 
 `content` 
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 任务因网络不通无法验证 SQL 及启动任务

问题描述：在使用 SQL 验证功能的时候会出现长时间卡住，最终出现 timeout 异常。导致无法通过 SQL 验证。如果对于线上任务，可能会出现长时间无法启动，最终失败的问题。

问题原因：在 SQL 验证和语法解析的过程中，会去访问 ByteHouse 服务端进行元数据获取。此时如果网络无法访问，则会出现长时间卡住的现象。

解决方案：

1. 在 ByteHouse·CDW 产品界面 租户管理 - 基本信息 - 网络信息，检查 Flink 任务运行的资源池的 VPC 和子网是否与 ByteHouse VPC 和子网相同：
   1. 如果处于不同 VPC 下，默认网络是不联通的，建议参考 创建资源池，创建一个与 ByteHouse VPC 相同的 Flink 资源池（建议）。或者采用 设置网络信息 方案设置公网访问域名，Flink 通过公网进行访问（不建议）。
   2. 如果处于相同 VPC 下
      1. 请检查安全组是否相同，如果不同请确认是否安全组的安全规则限制了 ByteHouse 的访问，如果有相关端口或者 IP 限制，请酌情放开限制。
      2. 请检查 Flink 与 ByteHouse 的子网是否相同，如果不同，请在私有网络 - 网络ACL 确认子网之间是否网络访问限制。如果有，请禁用相关限制。

导入 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 中增加 'timestamp-offset' = '-8h'，这样子在单个 ByteHouse Sink 中生效。
3. 推荐方案三：在 Flink 任务中统一设置变量，这个和方案二效果类似，但是对 Flink 任务中所有的 ByteHouse Sink 都生效containerized.taskmanager.env.FLINK_WRITE_TO_BYTEHOUSE_TIMESTAMP_OFFSET: -8h。