Flink Connector Driver for ByteHouse 云数仓版连接器专门用于通过 Flink 将数据加载到 ByteHouse云数仓版。本文将介绍通过 Table API&SQL 和 Flink 的 DataStreamAPI 两种方式连接ByteHouse并处理数据。

ByteHouse 支持通过 IAM 用户或数据库用户连接 Flink Connector Driver。IAM 用户与数据库用户二者差异说明如下，您可按需选择。

• IAM 用户为火山引擎访问控制（IAM）中创建的用户，其权限由 IAM 权限策略及您授予的 ByteHouse 资源和数据权限决定。IAM 用户可访问 ByteHouse 控制台，也支持通过 CLI、连接驱动、生态工具、API 等方式访问 ByteHouse。
• 数据库用户为 ByteHouse 中创建的数据库级别用户，可为其授予环境、资源和数据权限。数据库用户不可访问 ByteHouse 控制台，但支持通过 CLI、连接驱动、生态工具、API 等方式访问 ByteHouse。

更多 IAM 用户和数据库用户的介绍请参见以下文档：

• ByteHouse权限管控概述
• 管理 IAM 用户
• 管理数据库用户

根据您安装的 Flink 版本，下载匹配的 Flink SQL 或 Flink DataStream API 驱动。

Flink SQL 驱动

Flink 版本	驱动程序	发布日期
1.18	v2.12-1.27.117-1.18	2025-01-13
1.17	v2.12-1.27.117-1.17	2025-01-13
1.16	v2.12-1.27.117-1.16	2025-01-13
1.15	v2.12-1.27.117-1.15	2025-01-13
1.11 （scala 2.11）	flink-sql-connector-bytehouse-cdw_2.11-1.27.104-1.11.jar 未知大小	2024-10-23

本地安装

请运行以下命令，将下载的flink-sql-connector-bytehouse-cdw-${flink-sql-connector-bytehouse-cdw.version}.jar文件安装到本地 Maven 存储库：

mvn install:install-file \
    -Dfile=${your_path}/flink-sql-connector-bytehouse-cdw-${flink-sql-connector-bytehouse-cdw.version}.jar \
    -DgroupId=com.bytedance.bytehouse \
    -DartifactId=flink-sql-connector-bytehouse-cdw \
    -Dversion=${flink-sql-connector-bytehouse-cdw.version} \
    -Dpackaging=jar

然后，您可以使用 IDE（例如 IntelliJ）将其添加为项目中的依赖项。

Maven 依赖

对于要使用 Flink connector 连接器进行编译的 Maven 项目，请将以下依赖项添加到项目的 pom.xml 文件中。scala.version 需要是 2.11 或 2.12，与 Flink 发行版的关联 Scala 版本相对应。${flink-sql-connector-bytehouse-cdw.version} 可以设置为所需的 Flink connector 连接器版本。

<dependency>
    <groupId>com.bytedance.bytehouse</groupId>
    <artifactId>flink-sql-connector-bytehouse-cdw_${scala.version}</artifactId>
    <version>${flink-sql-connector-bytehouse-cdw.version}</version>
</dependency>

<dependency>
    <groupId>com.bytedance.bytehouse</groupId>
    <artifactId>flink-sql-connector-bytehouse-cdw_2.12</artifactId>
    <version>1.27.100_snapshot7</version>
</dependency>

然后，将以下存储库添加到 pom.xml文件：

<repository>
    <id>bytedance</id>
    <name>ByteDance Public Repository</name>
    <url>https://artifact.bytedance.com/repository/releases</url>
</repository>

通过 Flink SQL 导入

下面是通过 FlinkSQL 将数据表单加载到 ByteHouse 云数仓版数据表中的示例。

使用 IAM 用户连接并导入

您可参考步骤三：获取 ByteHouse 连接串信息中使用 IAM 用户连接所需的信息，替换下面对应的占位符。

• 'bytehouse.gateway.host'：配置为 ByteHouse 的公网/私网域名，您可以在 ByteHouse 控制台的租户管理 > 基本信息 > 网络信息中查看并复制网络信息。详情请参见步骤二：配置网络信息。
• 'bytehouse.gateway.api-token'：设置为 ByteHouse 的 <API_Key>，您可以在 ByteHouse 控制台的 租户管理 > 连接信息 中新建并复制 API Key。详情请参见获取 API Key。
• 'bytehouse.gateway.virtual-warehouse'：设置为 ByteHouse 计算组 ID，您可以在 ByteHouse 控制台的计算组 页面获取。
• 'database'：您需要连接的数据库名称。
• 'table-name'：您需要连接的表名称。

更多参数说明请参见参数说明。

CREATE TABLE `bh_de_source` ( -- 源表
 `id` BIGINT NOT NULL,
 `time` TIMESTAMP(0),
 `content` ARRAY<DECIMAL(20, 0)>
) WITH (
 'connector' = 'datagen'
);
CREATE TABLE `bh_de_sink` ( -- 目标表
 `id` STRING NOT NULL,
 `event_time` STRING,
 `content` ARRAY<DECIMAL(20, 0)>,
 PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
 'connector' = 'bytehouse-cdw',
 'jdbc.enable-gateway-connection' = 'true',
 'bytehouse.gateway.region' = 'VOLCANO_PRIVATE',
 'bytehouse.gateway.host' = 'tenant-{TENANT-ID}-{REGION}.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` ( -- 数据加载
 `id`,
 `event_time`,
 `content`
)
SELECT
 CAST(`id` AS STRING), -- 带有默认值的类型转换
 CAST(`time` AS STRING), -- 类型转换
 `content` -- 默认值
FROM `bh_de_source`;

使用数据库用户连接并导入

您可参考步骤三：获取 ByteHouse 连接串信息中使用数据库用户连接所需的信息，替换下面对应的占位符。更多参数说明请参见参数说明。

• 'bytehouse.gateway.host'：配置为 ByteHouse 的公网/私网域名，您可以在 ByteHouse 控制台的租户管理 > 基本信息 > 网络信息中查看并复制网络信息。详情请参见步骤二：配置网络信息。
• 'bytehouse.gateway.virtual-warehouse'：设置为 ByteHouse 计算组 ID，您可以在 ByteHouse 控制台的计算组 页面获取。
• 'bytehouse.gateway.account'：设置为火山引擎用户账号 ID 或名称，您可登录 ByteHouse 控制台，单击右上角个人中心查看并复制账号 ID 或名称。
• 'bytehouse.gateway.username'：设置为 ByteHouse 数据库账号用户名，您可登录 ByteHouse 控制台，在权限管理 > 用户页面中查看并复制数据库用户名。
• 'bytehouse.gateway.paasword'：设置为 ByteHouse 数据库用户的密码，该密码由管理员创建数据库账号时自定义配置，您可联系管理员获取密码。
• 'database'：您需要连接的数据库名称。
• 'table-name'：您需要连接的表名称。

CREATE TABLE `bh_de_source` ( -- 源表
 `id` BIGINT NOT NULL,
 `time` TIMESTAMP(0),
 `content` ARRAY<DECIMAL(20, 0)>
) WITH (
 'connector' = 'datagen'
);
CREATE TABLE `bh_de_sink` ( -- 目标表
 `id` STRING NOT NULL,
 `event_time` STRING,
 `content` ARRAY<DECIMAL(20, 0)>,
 PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
 'connector' = 'bytehouse-cdw',
 'jdbc.enable-gateway-connection' = 'true',
 'bytehouse.gateway.region' = 'VOLCANO_PRIVATE',
 'bytehouse.gateway.host' = 'tenant-{TENANT-ID}-{REGION}.bytehouse.ivolces.com',
 'bytehouse.gateway.port' = '19000',
 'bytehouse.gateway.virtual-warehouse' = '<VIRTUAL_WAREHOUSE>',
 'bytehouse.gateway.account' = '<your_volcano_engine_ID>',
 'username' = '<your_bytehouse_cdw_database_user>',
 'paasword' = '<your_bytehouse_cdw_database_user_password>',
 'database' = '<BYTEHOUSE_DB>',
 'table-name' = '<BYTEHOUSE_TABLE>'
);
INSERT INTO `bh_de_sink` ( -- 数据加载
 `id`,
 `event_time`,
 `content`
)
SELECT
 CAST(`id` AS STRING), -- 带有默认值的类型转换
 CAST(`time` AS STRING), -- 类型转换
 `content` -- 默认值
FROM `bh_de_source`;

通过 Flink DataStream API 导入

Flink 连接器的 DataStream API 将源数据类型限制为 RowData。
DataStream API 的使用主要是通过 CnchSinkFunctionBuilder 获取 CnchSinkFunction 实例，这有助于连接器的各种配置。
示例：多表数据库用户写入

使用 IAM 用户连接并导入

您可参考步骤三：获取 ByteHouse 连接串信息中使用 IAM 用户连接所需的信息，替换下面对应的占位符。更多参数说明请参见参数说明。

• .withGatewayConnection("VOLCANO_PRIVATE", "your_host", "your_port")：
  • "your_host"：配置为 ByteHouse 的公网/私网域名，您可以在 ByteHouse 控制台的租户管理 > 基本信息 > 网络信息中查看并复制网络信息。详情请参见步骤二：配置网络信息。
  • "your_port")：固定为 19000。
• .withGatewayApiToken("<< your API token >>")：设置为 ByteHouse 的 <API_Key>，您可以在 ByteHouse 控制台的 租户管理 > 连接信息 中获取API Key。详情请参见获取 API Key。
• .withGatewayVirtualWarehouse("default_vw")：设置为 ByteHouse 计算组 ID，您可以在 ByteHouse 控制台的计算组 页面获取。

import bytehouse.shaded.com.google.common.collect.Lists;
import com.bytedance.bytehouse.flink.connector.cnch.ByteHouseCdwClient;
import com.bytedance.bytehouse.flink.connector.cnch.CnchMultiSinkFunction;
import com.bytedance.bytehouse.flink.connector.cnch.api.java.CnchMultiSinkFunctionBuilder;
import com.bytedance.bytehouse.flink.connector.cnch.sql.ByteHouseCdwCreateTableStatementBuilder;
import com.bytedance.bytehouse.table.ByteHouseColumn;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public final class MultiSinkCdwTableExample {
  public static void main(String[] args) throws Exception {
    final String region = "VOLCANO_PRIVATE";
    final String host = args[0]; // e.g., tenant-{accountId}-{region}-public.bytehouse.volces.com
    final int port = 19000;
    final String token = args[1]; // e.g., 1HkKORvYGV:xxxxxxxx
    final String vwId = args[2]; // e.g., vw-{environment_id}{account_id}-{virtual_warehouse_name}
    final String dbName = args[3];
    final String tableName = args[4];
    List<RowData> collectedRowData = Lists.newArrayList();
    ByteHouseCdwClient bhCli = new ByteHouseCdwClient(region, host, port, token, vwId);
    // Drop and recreate tables
    final String tableName1 = tableName + "1";
    bhCli.dropTableIfExists(dbName, tableName1);
    bhCli.execute(
        new ByteHouseCdwCreateTableStatementBuilder()
            .createTable(dbName, tableName1)
            .schema("id Int64, event_time DateTime, content String")
            .partitionBy("toYYYYMMDD(`event_time`)")
            .orderBy("id")
            .uniqueKey("id")
            .build());
    final String tableName2 = tableName + "2";
    bhCli.dropTableIfExists(dbName, tableName2);
    bhCli.execute(
        new ByteHouseCdwCreateTableStatementBuilder()
            .createTable(dbName, tableName2)
            .schema("id Int64, properties Map(String, String)")
            .addColumn(ByteHouseColumn.of("login_time", "Nullable(Time(3))"))
            .build());
    // Construct data
    for (int i = 0; i < 10; i++) {
      Map<String, Object> record = new HashMap<>();
      record.put("id", i);
      record.put("event_time", LocalDateTime.now());
      record.put("login_time", LocalTime.now());
      record.put("content", String.format("Row %d @ %s", i, record.get("event_time")));
      Map<String, String> props = new HashMap<>();
      props.put("propId", record.get("id").toString());
      props.put("propEventTime", record.get("event_time").toString());
      props.put("propContent", record.get("content").toString());
      record.put("properties", props);
      RowData rowData1 = bhCli.constructRichRowData(RowKind.INSERT, dbName, tableName1, record);
      collectedRowData.add(rowData1);
      RowData rowData2 = bhCli.constructRichRowData(RowKind.INSERT, dbName, tableName2, record);
      collectedRowData.add(rowData2);
    }
    // Create the execution environment
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // Add a source to the data stream
    DataStream<RowData> dataStream = env.fromCollection(collectedRowData);
    try (CnchMultiSinkFunction<RowData> sink =
        new CnchMultiSinkFunctionBuilder.MultiRowDataTable(Duration.ofMinutes(3))
            .withGatewayConnection(region, host, port)
            .withGatewayApiToken(token)
            .withGatewayVirtualWarehouse(vwId)
            .withFlushInterval(Duration.ofSeconds(10))
            .withFlushBatchSize(5000)
            .build()) {
      // Add a sink to the data stream
      dataStream.addSink(sink);
      // Trigger the execution
      env.execute(MultiSinkCdwTableExample.class.getSimpleName());
    }
    // Select and print out all rows in the table
    System.out.printf("\n==== %s ====\n", tableName1);
    bhCli.selectAll(dbName, tableName1).forEach(System.out::println);
    System.out.printf("\n==== %s ====\n", tableName2);
    bhCli.selectAll(dbName, tableName2).forEach(System.out::println);
    // Clean up
    bhCli.close();
  }
}

使用数据库用户连接并导入

您可参考步骤三：获取 ByteHouse 连接串信息中使用数据库用户连接所需的信息，替换下面对应的占位符。更多参数说明请参见参数说明。

• .withGatewayConnection("VOLCANO_PRIVATE", "your_host", "your_port")：
  • "your_host"：配置为 ByteHouse 的公网/私网域名，您可以在 ByteHouse 控制台的租户管理 > 基本信息 > 网络信息中查看并复制网络信息。详情请参见步骤二：配置网络信息。
  • "your_port")：固定为 19000。
• .withGatewayAccount(accountId, username, password)：
  • accountId：设置为火山引擎用户账号 ID 或名称，您可登录 ByteHouse 控制台，单击右上角个人中心查看并复制账号 ID 或名称。
  • username：设置为 ByteHouse 数据库账号用户名，您可登录 ByteHouse 控制台，在权限管理 > 用户页面中查看并复制数据库用户名。
  • password：设置为 ByteHouse 数据库用户的密码，该密码由管理员创建数据库账号时自定义配置，您可联系管理员获取密码。
• .withGatewayVirtualWarehouse("default_vw")：设置为 ByteHouse 计算组 ID，您可以在 ByteHouse 控制台的计算组 页面获取。

import bytehouse.shaded.com.google.common.collect.Lists;
import com.bytedance.bytehouse.flink.connector.cnch.ByteHouseCdwClient;
import com.bytedance.bytehouse.flink.connector.cnch.CnchMultiSinkFunction;
import com.bytedance.bytehouse.flink.connector.cnch.api.java.CnchMultiSinkFunctionBuilder;
import com.bytedance.bytehouse.flink.connector.cnch.sql.ByteHouseCdwCreateTableStatementBuilder;
import com.bytedance.bytehouse.table.ByteHouseColumn;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public final class MultiSinkCdwTableExample {
  public static void main(String[] args) throws Exception {
    final String region = "VOLCANO_PRIVATE";
    final String host = args[0]; // e.g., tenant-2100000000-cn-guangzhou-public.bytehouse.volces.com
    final int port = 19000;
    final String accountId = args[1];
    final String username = args[2];
    final String password = args[3];
    final String vwId = args[4]; // e.g., vw-2100000000-default-vw
    final String dbName = args[5];
    final String tableName = args[6];
    List<RowData> collectedRowData = Lists.newArrayList();
    ByteHouseCdwClient bhCli = new ByteHouseCdwClient(region, host, port, accountId, username, password, vwId);
    // Drop and recreate tables
    final String tableName1 = tableName + "1";
    bhCli.dropTableIfExists(dbName, tableName1);
    bhCli.execute(
        new ByteHouseCdwCreateTableStatementBuilder()
            .createTable(dbName, tableName1)
            .schema("id Int64, event_time DateTime, content String")
            .partitionBy("toYYYYMMDD(`event_time`)")
            .orderBy("id")
            .uniqueKey("id")
            .build());
    final String tableName2 = tableName + "2";
    bhCli.dropTableIfExists(dbName, tableName2);
    bhCli.execute(
        new ByteHouseCdwCreateTableStatementBuilder()
            .createTable(dbName, tableName2)
            .schema("id Int64, properties Map(String, String)")
            .addColumn(ByteHouseColumn.of("login_time", "Nullable(Time(3))"))
            .build());
    // Construct data
    for (int i = 0; i < 10; i++) {
      Map<String, Object> record = new HashMap<>();
      record.put("id", i);
      record.put("event_time", LocalDateTime.now());
      record.put("login_time", LocalTime.now());
      record.put("content", String.format("Row %d @ %s", i, record.get("event_time")));
      Map<String, String> props = new HashMap<>();
      props.put("propId", record.get("id").toString());
      props.put("propEventTime", record.get("event_time").toString());
      props.put("propContent", record.get("content").toString());
      record.put("properties", props);
      RowData rowData1 = bhCli.constructRichRowData(RowKind.INSERT, dbName, tableName1, record);
      collectedRowData.add(rowData1);
      RowData rowData2 = bhCli.constructRichRowData(RowKind.INSERT, dbName, tableName2, record);
      collectedRowData.add(rowData2);
    }
    // Create the execution environment
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // Add a source to the data stream
    DataStream<RowData> dataStream = env.fromCollection(collectedRowData);
    try (CnchMultiSinkFunction<RowData> sink =
        new CnchMultiSinkFunctionBuilder.MultiRowDataTable(Duration.ofMinutes(3))
            .withGatewayConnection(region, host, port)
            .withGatewayAccount(accountId, username, password)
            .withGatewayVirtualWarehouse(vwId)
            .withFlushInterval(Duration.ofSeconds(10))
            .withFlushBatchSize(5000)
            .build()) {
      // Add a sink to the data stream
      dataStream.addSink(sink);
      // Trigger the execution
      env.execute(MultiSinkCdwTableExample.class.getSimpleName());
    }
    // Select and print out all rows in the table
    System.out.printf("\n==== %s ====\n", tableName1);
    bhCli.selectAll(dbName, tableName1).forEach(System.out::println);
    System.out.printf("\n==== %s ====\n", tableName2);
    bhCli.selectAll(dbName, tableName2).forEach(System.out::println);
    // Clean up
    bhCli.close();
  }
}