ByteHouse 企业版 Spark Connector 连接器专门用于通过 Spark 将数据加载到 ByteHouse 企业版。
本文将介绍通过 Spark SQL ，以及 EMR 支持的 Servless Spark 两种方式连接ByteHouse并处理数据。

先决条件

• Java8
• Spark 3.5 及以上版本

驱动安装

请按照下面的方法，在程序中配置以下驱动的依赖项。

1. ByteHouse Spark Connector。
2. ClickHouse JDBC 驱动程序。
3. ByteHouse JDBC 驱动程序。

Maven 依赖

对于要使用 Spark connector 连接器进行编译的 Maven 项目，请将以下依赖项添加到项目的 pom.xml 文件中。

<dependency>
    <groupId>com.clickhouse</groupId>
    <artifactId>clickhouse-jdbc</artifactId>
    <version>0.4.6</version>
</dependency>
<dependency>
    <groupId>com.bytedance.bytehouse-ce</groupId>
    <artifactId>clickhouse-spark-runtime-3.5_2.12</artifactId>
    <version>0.8.1.1</version>
</dependency>

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

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

用法示例

Spark SQL

您可以参考下面的命令，基于 Spark SQL CLI 连接到 ByteHouse。
请注意替换：

1. 填写 export 字段中的 CLICKHOUSE_HOST和 CLICKHOUSE_PASSWORD 的值，您可以参见 获取 ByteHouse 连接信息来获取；
2. 填写 export 字段中的数据库名CLICKHOUSE_DATABASE和计算组名CLICKHOUSE_VW。
3. --jars 配置中实际使用的 jar 实际文件路径。

export SPARK_LOCAL_IP=localhost
export SPARK_HOME=/opt/tiger/spark
export CLICKHOUSE_HOST=<your-host>
export CLICKHOUSE_HTTP_PORT=8123
export CLICKHOUSE_USER=<your-username>
export CLICKHOUSE_PASSWORD='<your-password>'

$SPARK_HOME/bin/spark-sql \
  --conf spark.sql.catalog.clickhouse=xenon.clickhouse.ClickHouseCatalog \
  --conf spark.sql.catalog.clickhouse.host=${CLICKHOUSE_HOST:-127.0.0.1} \
  --conf spark.sql.catalog.clickhouse.protocol=http \
  --conf spark.sql.catalog.clickhouse.http_port=${CLICKHOUSE_HTTP_PORT:-8123} \
  --conf spark.sql.catalog.clickhouse.user=${CLICKHOUSE_USER:-default} \
  --conf spark.sql.catalog.clickhouse.password=${CLICKHOUSE_PASSWORD:-} \
  --conf spark.sql.catalog.clickhouse.database=default \
  --conf spark.sql.catalog.clickhouse.shard-discovery.kind=CE_API_CLUSTERS \
  --conf spark.sql.catalog.clickhouse.bytehouse-ce.api.account-id=xx \
  --conf spark.sql.catalog.clickhouse.clickhouse.cluster=xx \
  --conf spark.driver.bindAddress=127.0.0.1 \
  --conf spark.clickhouse.write.distributed.convertLocal=true \
  --conf spark.clickhouse.write.useShardsWriter=true \
  --conf spark.clickhouse.write.format=json \
  --conf spark.clickhouse.read.filterByPartition=false \
  --jars clickhouse-jdbc-0.4.6-all.jar,clickhouse-spark-runtime-3.5_2.12-0.8.1.1.jar

Servless Spark

Servless Spark 方式适用于火山引擎 EMR（E-MapReduce）服务。通过此方式连接，您需要将driver下载并上传到 TOS 中调用。
您可以参考下面的命令，基于 Servless Spark 方式连接到 ByteHouse。
请注意替换：

1. 填写set 中的 vpc 网络信息。您可以在 ByteHouse控制台 集群管理-集群详情-基本信息 页面查询到此信息。
2. 填写set 中的 AK/SK（access key、secret key）信息，您可参见 获取Access Key。
3. 填写 set 字段中的公网或私网 HOST 地址，以及账户密码信息，您可以参见 获取集群连接信息；同时设置 database 数据库名和VW 计算组名。
4. 填写实际使用的驱动 jar 文件的 TOS 路径。

set serverless.spark.analysis=true;
set spark.hadoop.fs.tos.skip.resolve = true;
set las.cross.vpc.access.enabled = true;
set las.cross.vpc.vpc.id = <your-vpc-id>;
-- 使用可用区B子网
set las.cross.vpc.subnet.id = <your-subnet-id>;
set las.cross.vpc.security.group.id = <your-securitygroup-id>;
set las.cross.vpc.accountId=<your-account-id>;
set las.cluster.type = vke;
set emr.serverless.spark.only.parse.enabled  = true;
set serverless.spark.access.key=<your-tos-ak>;
set serverless.spark.secret.key=<your-tos-sk>;
set spark.sql.catalog.clickhouse=xenon.clickhouse.ClickHouseCatalog;
set spark.sql.catalog.clickhouse.host=<host>;
set spark.sql.catalog.clickhouse.protocol=http;
set spark.sql.catalog.clickhouse.http_port=8123;
set spark.sql.catalog.clickhouse.user=<your-username>;
set spark.sql.catalog.clickhouse.password=<your-password>;
set spark.sql.catalog.clickhouse.database=<your-database>;
set spark.sql.catalog.clickhouse.shard_discovery.kind=CE_API_CLUSTERS;
set spark.sql.catalog.clickhouse.bytehouse_ce.api.account_id=<your-account-id>;
set spark.sql.catalog.clickhouse.cluster=<your-cluster-name>;
set spark.sql.catalog.clickhouse.option.socket_timeout=300000;
set spark.sql.catalog.clickhouse.option.connect_timeout=30000;
set spark.sql.catalog.clickhouse.option.custom_settings=max_execution_time=3000;
set spark.clickhouse.write.format=json;
set spark.clickhouse.read.filterByPartition=false;
set spark.clickhouse.write.distributed.convertLocal=true;
set spark.clickhouse.write.useShardsWriter=true;

set las.spark.jar.depend.jars = [{"schema":"","fileName":"tos://test-bh-tos/clickhouse-jdbc-0.4.6-all.jar"},{"schema":"","fileName":"tos://test-bh-tos/clickhouse-spark-runtime-3.5_2.12-0.8.1.1.jar"}];

use clickhouse;

insert into people_t_copy select * from people_t; 

配置参数

支持场景

支持的数据类型