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

准备工作

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

Flink SQL 驱动

Flink 版本	驱动程序	发布日期
1.18	flink-sql-connector-bytehouse-cdw_2.12-1.27.117-1.18.jar 未知大小	2025-01-13
1.17	flink-sql-connector-bytehouse-cdw_2.12-1.27.117-1.17.jar 未知大小	2025-01-13
1.16	flink-sql-connector-bytehouse-cdw_2.12-1.27.117-1.16.jar 未知大小	2025-01-13
1.15	flink-sql-connector-bytehouse-cdw_2.12-1.27.117-1.15.jar 未知大小	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 云数仓版数据表中的示例。

CREATE TABLE `bh_de_source` (  -- source table
  `id` BIGINT NOT NULL,
  `time` TIMESTAMP(0),
  `content` ARRAY<DECIMAL(20, 0)>
)
PARTITIONED BY (`id`)
WITH (
  'connector' = 'kinesis',
  'stream' = 'demo_stream',
  'format' = 'json',
  'aws.region' = 'cn-north-1',
  'aws.credentials.provider' = 'BASIC',
  'aws.credentials.basic.accesskeyid' = '???',
  'aws.credentials.basic.secretkey' = '???',
  'scan.shard.getrecords.maxretries' = '7'
);

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`

通过 Flink DataStream API 导入

Flink 连接器的 DataStream API 将源数据类型限制为 RowData。
DataStream API 的使用主要是通过 CnchSinkFunctionBuilder 获取 CnchSinkFunction 实例，这有助于连接器的各种配置。下面是一个演示基本用法的示例。

import com.bytedance.bytehouse.flink.connector.cnch.CnchSinkFunction;
import com.bytedance.bytehouse.flink.connector.cnch.api.java.CnchSinkFunctionBuilder;
import com.bytedance.bytehouse.flink.table.data.RowDataConstructor;
import java.sql.Date;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;

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();
}

参数说明