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

下载驱动

• Spark Connector

  spark 版本	驱动程序	发布日期
  3.5及以上版本	clickhouse-spark-runtime-3.5_2.12-0.8.1.1.jar 未知大小	2024-09-11

• ByteHouse JDBC
  

  

  clickhouse-jdbc-0.4.6.jar

  未知大小

  

添加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。

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 方式连接到 ByteHouse。Servless Spark 方式适用于火山引擎 EMR（E-MapReduce）服务。

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=xx.bytehouse-ce.ivolces.com;
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"}];

-- 从 hive 的 people_t 导入 bh 的 people_t_copy 表
use clickhouse;
insert into people_t_copy select * from spark_catalog.default.people_t; 

其中关键连接参数为：

配置参数

数据类型映射

支持的Spark Catalog函数

CREATE TABLE [IF NOT EXISTS] [tableIdentifier] [UUID uuid]
(
    [tableColumnDfnt],
    [CONSTRAINT constraint_name CHECK columnExpr,]
    ...
) [engineClause]