The ByteHouse Cloud Data Warehouse Edition Spark Connector is specifically designed for loading data into ByteHouse Cloud Data Warehouse Edition via Spark. This article introduces three methods for connecting to ByteHouse and processing data: Spark SQL, EMR-supported Serverless Spark, and Spark Jar.

• Java 8
• Scala 2.12 and later
• Spark 3.5

Please configure the dependencies for the following drivers in the program according to the method below.

1. ByteHouse Spark Connector.
2. The ClickHouse JDBC driver.
3. ByteHouse JDBC driver.

Maven dependencies #

For Maven projects that need to be compiled using the Spark connector, add the following dependencies to the project's pom.xml file.

1. Add the following dependencies.

   Spark version	Dependencies	Download
   Spark 3.5	<dependency> <groupId>com.clickhouse</groupId> <artifactId>clickhouse-jdbc</artifactId> <version>0.4.6</version> </dependency> <dependency> <groupId>com.bytedance.bytehouse</groupId> <artifactId>driver-java</artifactId> <version>1.1.75</version> <classifier>all</classifier> </dependency> <dependency> <groupId>com.bytedance.bytehouse</groupId> <artifactId>clickhouse-spark-runtime-3.5_2.12</artifactId> <version>0.8.0.15</version> </dependency>	clickhouse-jdbc-0.4.6.jar Unknown size driver-java-1.1.75-all.jar Unknown size clickhouse-spark-runtime-3.5_2.12-0.8.0.15.jar Unknown size

2. Add the following repositories to the pom.xml file:

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

Guide to configuring connection parameters #

The general parameters and configuration instructions required to connect to ByteHouse are as follows. For more information about configuring connection parameters, see the section below 配置参数.

Local testing #

Before running local tests, please load the following JAR files into ${SPARK_HOME/jars}, or specify the local path to the JAR files using --jars when submitting the command.

1. ByteHouse Spark Connector。
2. ClickHouse JDBC driver.
3. ByteHouse JDBC driver.

Spark SQL #

Replace CLICKHOUSE_HOST, CLICKHOUSE_PASSWORD, CLICKHOUSE_DATABASE, and CLICKHOUSE_VW with actual values. For information on how to obtain these values, see 连接参数配置说明.

export SPARK_LOCAL_IP=localhost
export SPARK_HOME=/opt/tiger/spark
export CLICKHOUSE_HOST=bytehouse-<REGION>.volces.com
export CLICKHOUSE_PASSWORD=<your-api-key>
export CLICKHOUSE_DATABASE=<database>
export CLICKHOUSE_VW=<your-virtual-warehouse>

$SPARK_HOME/bin/spark-sql \
  --conf spark.sql.catalog.clickhouse=com.bytehouse.ByteHouseCatalog \
  --conf spark.sql.catalog.clickhouse.host=${CLICKHOUSE_HOST:-127.0.0.1} \
  --conf spark.sql.catalog.clickhouse.password=${CLICKHOUSE_PASSWORD:-} \
  --conf spark.sql.catalog.clickhouse.database=${CLICKHOUSE_DATABASE} \
  --conf spark.sql.catalog.clickhouse.option.vw=${CLICKHOUSE_VW} \
  --conf spark.driver.bindAddress=127.0.0.1 \
  --conf spark.bytehouse.write.batchSize=50000 \
  --conf spark.bytehouse.write.format=jdbc \
  --jars clickhouse-jdbc-0.4.6-all.jar,driver-java-1.1.75-all.jar,clickhouse-spark-runtime-3.5_2.12-0.8.0.15.jar

Spark Jar #

When using, replace spark.sql.catalog.clickhouse.host, spark.sql.catalog.clickhouse.password, spark.sql.catalog.clickhouse.database, and spark.sql.catalog.clickhouse.option.vw with actual values. For information on how to obtain these values, see 连接参数配置说明.

import org.apache.spark.sql.SparkSession

object Main {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("App")
      .master("local")
      .config("spark.sql.catalog.clickhouse", "com.bytehouse.ByteHouseCatalog")
      .config("spark.sql.catalog.clickhouse.host", "<your-host>")
      .config("spark.sql.catalog.clickhouse.protocol", "http")
      .config("spark.sql.catalog.clickhouse.password", "<your-api-key>")
      .config("spark.sql.catalog.clickhouse.database", "<database>")
      .config("spark.sql.catalog.clickhouse.option.vw", "<your-virtual-warehouse>")
      .config("spark.sql.catalog.clickhouse.option.query_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.connect_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.receive_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.send_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.max_execution_time", "108000000")
      .config("spark.bytehouse.write.batchSize", "50000")
      .config("spark.bytehouse.write.format", "jdbc")
      .getOrCreate()

    val sql = "select * from clickhouse.xx.xx"
    val df = spark.sql(sql)
    spark.stop()
  }
}

PySpark #

When using, replace spark.sql.catalog.clickhouse.host, spark.sql.catalog.clickhouse.password, spark.sql.catalog.clickhouse.database, and spark.sql.catalog.clickhouse.option.vw with actual values. For information on how to obtain these values, see 连接参数配置说明.

from pyspark.sql import SparkSession
spark = SparkSession\
    .builder\
    .appName("App")\
    .master("local")\
    .config("spark.sql.catalog.clickhouse", "com.bytehouse.ByteHouseCatalog")\
    .config("spark.sql.catalog.clickhouse.host", "<your-host>")\
    .config("spark.sql.catalog.clickhouse.protocol", "http")\
    .config("spark.sql.catalog.clickhouse.password", "<your-api-key>")\
    .config("spark.sql.catalog.clickhouse.database", "<database>")\
    .config("spark.sql.catalog.clickhouse.option.vw", "<your-virtual-warehouse>")\
    .config("spark.sql.catalog.clickhouse.option.query_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.connect_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.receive_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.send_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.max_execution_time", "108000000")\
    .config("spark.bytehouse.write.batchSize", "50000")\
    .config("spark.bytehouse.write.format", "jdbc")\
    .config("spark.jars", "clickhouse-jdbc-0.4.6-all.jar,driver-java-1.1.75-all.jar,clickhouse-spark-runtime-3.5_2.12-0.8.0.15.jar").\
    getOrCreate()
sql = "select count(*) from clickhouse.xxx.xxx"
df = spark.sql(sql)
df.show()
spark.stop()

Serverless Spark #

The Serverless Spark method is applicable to the Volcengine EMR (E-MapReduce) service. You can refer to the following commands to connect to ByteHouse using the Serverless Spark method. For Serverless Spark job development, please refer to relevant documentation, such as the Serverless Spark Job Development Guide.
You can connect ByteHouse and Volcano Engine EMR (E-MapReduce) services using Serverless Spark, which supports connections via Spark SQL, Spark Jar, and PySpark.

Spark SQL #

You can refer to the following command to connect to ByteHouse using Spark SQL CLI. For details, see Spark SQL Job Development Guide.

Use the default compute group to connect

The following command uses the Spark SQL default compute group to connect to ByteHouse. When using the command, replace the placeholders with the actual parameter values. For information on how to obtain these values, see the parameter descriptions below.

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=com.bytehouse.ByteHouseCatalog;
set spark.sql.catalog.clickhouse.host=tenant-<your-account-id>-<REGION>.bytehouse.ivolces.com;
set spark.sql.catalog.clickhouse.protocol=http;
set spark.sql.catalog.clickhouse.password=<your-api-key>;
set spark.sql.catalog.clickhouse.database=<database>;
set spark.sql.catalog.clickhouse.option.vw=<your-virtual-warehouse>;

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

-- 查询 hive
-- select * from spark_catalog.hive_db.hive_table;
-- select * from people_t_copy;
insert into people_t_copy select * from people_t; 

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=com.bytehouse.ByteHouseCatalog;
set spark.sql.catalog.clickhouse.host=bytehouse-<REGION>.volces.com;
set spark.sql.catalog.clickhouse.protocol=http;
set spark.sql.catalog.clickhouse.password=<your-api-key>;
set spark.sql.catalog.clickhouse.database=<database>;
set spark.sql.catalog.clickhouse.option.vw=<your-virtual-warehouse>;

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

-- 查询 hive
-- select * from spark_catalog.hive_db.hive_table;
-- select * from people_t_copy;
insert into people_t_copy select * from people_t; 

Parameter description

Use Spark SQL to perform a group join

If you have created a Spark SQL compute group and want to use it to connect to ByteHouse, you can do so as follows. For an introduction to Spark SQL compute groups, see Queue Management.

1. Log in to the Volcengine EMR console. Under the Resource Management tab, click Serverless; click the queue you are using; click Compute Group; click Spark SQL Compute Group; click Parameter Configuration; then configure the following spark.jars custom parameters.
   

   {
     "spark.jars": "tos://bytehouse-spark/cdw/driver-java-1.1.75-all.jar,tos://bytehouse-spark/cdw/clickhouse-jdbc-0.4.6-all.jar,tos://bytehouse-spark/cdw/clickhouse-spark-runtime-3.5_2.12-0.8.0.15.jar"
   }
   

2. After configuration is complete, you can use the following commands to connect to ByteHouse. When using the command, replace the placeholders with the actual parameter values. For information on how to obtain these values, refer to the parameter description below.

   Private network connection

   Public network connection

   set serverless.spark.analysis=true;
      set spark.hadoop.fs.tos.skip.resolve = true;
      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=com.bytehouse.ByteHouseCatalog;
      set spark.sql.catalog.clickhouse.host=tenant-<your-account-id>-<REGION>.bytehouse.ivolces.com;
      set spark.sql.catalog.clickhouse.protocol=http;
      set spark.sql.catalog.clickhouse.password=<your-api-key>;
      set spark.sql.catalog.clickhouse.database=<database>;
      set spark.sql.catalog.clickhouse.option.vw=<your-virtual-warehouse>;
      
      use clickhouse;
      
      -- 查询 hive
      -- select * from spark_catalog.hive_db.hive_table;
      -- select * from people_t_copy;
      insert into people_t_copy select * from people_t; 
   

   set serverless.spark.analysis=true;
      set spark.hadoop.fs.tos.skip.resolve = true;
      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=com.bytehouse.ByteHouseCatalog;
      set spark.sql.catalog.clickhouse.host=bytehouse-<REGION>.volces.com;
      set spark.sql.catalog.clickhouse.protocol=http;
      set spark.sql.catalog.clickhouse.password=<your-api-key>;
      set spark.sql.catalog.clickhouse.database=<database>;
      set spark.sql.catalog.clickhouse.option.vw=<your-virtual-warehouse>;
      
      use clickhouse;
      
      -- 查询 hive
      -- select * from spark_catalog.hive_db.hive_table;
      -- select * from people_t_copy;
      insert into people_t_copy select * from people_t; 
   

   Parameter description

   Parameters	Configuration instructions
   serverless.spark.access.key、serverless.spark.secret.key	Configure to use AK/SK (access key, secret key). You can refer to Obtain an Access Key.
   spark.sql.catalog.clickhouse	host: Configure the private or public HOST address for ByteHouse. You can refer to 步骤二：配置网络信息 to obtain it. password: Configure the ByteHouse API key. You can refer to 获取 API Key to obtain it. Configure the ByteHouse database name. You can view and copy the name of the database to connect to through ByteHouse Console > Database. vw: Configured as the compute group name for ByteHouse. You can view the default compute group via ByteHouse Console > Tenant Management > Parameter Settings.
   las.spark.jar.depend.jars	Configure the TOS path to the jar that is actually used.
   Other	For more information about configuring connection parameters, see the section below 配置参数.

   warning

   The expected format for the set parameter is "SET", "SET key", or "SET key=value". If a key needs to contain special characters, or a value contains a semicolon (;), you can enclose it in backticks (````).

Spark Jar #

You can refer to the following commands to connect to ByteHouse using the Spark Jar method. Spark Jar job development guide: For details, see Spark Jar job development guide.
When using, replace spark.sql.catalog.clickhouse.host, spark.sql.catalog.clickhouse.password, spark.sql.catalog.clickhouse.database, and spark.sql.catalog.clickhouse.option.vw with actual values. For information on how to obtain these values, see 连接参数配置说明.

import org.apache.spark.sql.SparkSession

object Main {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("App")
      .master("local")
      .config("spark.sql.catalog.clickhouse", "com.bytehouse.ByteHouseCatalog")
      .config("spark.sql.catalog.clickhouse.host", "<your-host>")
      .config("spark.sql.catalog.clickhouse.protocol", "http")
      .config("spark.sql.catalog.clickhouse.password", "<your-api-key>")
      .config("spark.sql.catalog.clickhouse.database", "<database>")
      .config("spark.sql.catalog.clickhouse.option.vw", "<your-virtual-warehouse>")
      .config("spark.sql.catalog.clickhouse.option.query_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.connect_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.receive_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.send_timeout", "600000")
      .config("spark.sql.catalog.clickhouse.option.max_execution_time", "108000000")
      .config("spark.bytehouse.write.batchSize", "50000")
      .config("spark.bytehouse.write.format", "jdbc")
      .getOrCreate()

    val sql = "select * from clickhouse.xx.xx"
    val df = spark.sql(sql)
    spark.stop()
  }
}

PySpark #

You can refer to the following command to connect to ByteHouse using PySpark. For details on the PySpark Job Development Guide, see PySpark Job Development Guide.
When using, replace spark.sql.catalog.clickhouse.host, spark.sql.catalog.clickhouse.password, spark.sql.catalog.clickhouse.database, and spark.sql.catalog.clickhouse.option.vw with actual values. For information on how to obtain these values, see 连接参数配置说明.

from pyspark.sql import SparkSession
spark = SparkSession\
    .builder\
    .appName("App")\
    .master("local")\
    .config("spark.sql.catalog.clickhouse", "com.bytehouse.ByteHouseCatalog")\
    .config("spark.sql.catalog.clickhouse.host", "<your-host>")\
    .config("spark.sql.catalog.clickhouse.protocol", "http")\
    .config("spark.sql.catalog.clickhouse.password", "<your-api-key>")\
    .config("spark.sql.catalog.clickhouse.database", "<database>")\
    .config("spark.sql.catalog.clickhouse.option.vw", "<your-virtual-warehouse>")\
    .config("spark.sql.catalog.clickhouse.option.query_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.connect_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.receive_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.send_timeout", "600000")\
    .config("spark.sql.catalog.clickhouse.option.max_execution_time", "108000000")\
    .config("spark.bytehouse.write.batchSize", "50000")\
    .config("spark.bytehouse.write.format", "jdbc")\
    getOrCreate()
sql = "select count(*) from clickhouse.xxx.xxx"
df = spark.sql(sql)
df.show()
spark.stop()

Configuration parameters #

Configure driver parameters #

Set timeout parameters #

You can set the timeout period: using the following parameters.

set spark.sql.catalog.clickhouse.option.socket_timeout=60000;
set spark.sql.catalog.clickhouse.option.query_timeout=60000;
set spark.sql.catalog.clickhouse.option.connect_timeout=60000;
set spark.sql.catalog.clickhouse.option.receive_timeout=60000;
set spark.sql.catalog.clickhouse.option.send_timeout=60000;
set spark.sql.catalog.clickhouse.option.max_execution_time=10800000;

Other driver custom parameters #

If you need to add custom parameters, you can do so by following the spark.sql.catalog.clickhouse.option.xxx pattern.

set spark.sql.catalog.clickhouse.option.xxx=xxx;

Parallel read #

Spark supports parallel reading of bucket tables and partitioned tables, and also allows both bucket parallel reading and partition parallel reading to be enabled simultaneously, with the overall parallelism equal to the product of the two.

Bucket tables #

If the table being read is a bucket table, parallel reading by bucket can be enabled, with the degree of parallelism equal to the number of buckets. The table creation statement in this example specifies a parallelism of 32.

CREATE TABLE IF NOT EXISTS test.createTable(
id UInt32,
name String DEFAULT ''
) 
ENGINE=CnchMergeTree
ORDER BY id
CLUSTER BY EXPRESSION cityHash64(id) % 32 INTO 32 BUCKETS

# 开启按 bucket 并行读
set spark.bytehouse.read.byBucket=true;

Partition table #

If the table being read is a partitioned table, parallel reading by partition can be enabled.
An example of a table creation statement is as follows:

CREATE TABLE IF NOT EXISTS test.createTable(
id UInt32,
name String DEFAULT '',
date Date
) 
ENGINE=CnchMergeTree
ORDER BY id
PARTITION BY date

Suppose you need to query data for date within the time range [2025-01-01, 2025-01-10]; you can use the following configuration with a parallelism of 10.

set spark.bytehouse.read.byPartition=true;
set spark.bytehouse.read.partition.lower=20250101;
set spark.bytehouse.read.partition.upper=20250110;

Write to a bucket #

Spark supports writing to bucket tables. The following example demonstrates how to create the test.createTable table and enable data writing by configuring driver parameters.

CREATE TABLE IF NOT EXISTS test.createTable(
id UInt32,
name String DEFAULT ''
) 
ENGINE=CnchMergeTree
ORDER BY id
CLUSTER BY EXPRESSION cityHash64(id) % 32 INTO 32 BUCKETS

set spark.bytehouse.write.sharding-strategy=HASH;
set spark.bytehouse.write.sharding-key=id;
set spark.bytehouse.write.sharding-expression=`cityHash64(id) % 32`;

SHOW TABLES FROM `xx`

SHOW TABLES FROM `xx` LIKE `xxx`

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

DROP TABLE `xx`.`xxx`

RENAME TABLE `xx`.`xxx` to `xx`.`xxx`

SHOW DATABASES

CREATE DATABASE `xx` [with properties ('engine'='cnch')]

DROP DATABASE `xx`