Spark ClickHouse Connector

When joining two large tables, Spark typically performs a Sort Merge Join, which requires an expensive shuffle and sort by the join key.

If the ClickHouse tables are already collocated by their join keys (using a layout compatible with Storage-Partitioned Join or V2 Bucket Join), the connector can leverage this existing ClickHouse table layout. This allows Spark to perform a Bucket Join, eliminating the need for the expensive shuffle and sort operations typically required by a standard Sort Merge Join.

When writing to a Distributed table with spark.clickhouse.write.distributed.convertLocal=true, Spark bypasses ClickHouse's native routing and evaluates the sharding key itself. If the sharding key uses complex expressions (e.g., cityHash64(col_1, col_2)) that Spark cannot currently process, you must manage the risk of data corruption.

By default, the connector prevents silent corruption by throwing an error if unsupported expressions are detected. To allow these writes, you must explicitly set:

1. spark.clickhouse.ignoreUnsupportedTransform to true (to ignore the error and log a warning).
2. spark.clickhouse.write.distributed.convertLocal.allowUnsupportedSharding to true (to allow the write to proceed).

Warning: This combination is dangerous and may cause incorrect data distribution. Only use this if you have verified your data distribution manually.

The connector supports Spark's DataSource V2 push down capabilities, allowing Spark to delegate query processing directly to ClickHouse. This reduces network traffic, improves performance, and lowers the load on ClickHouse by performing operations at the source rather than pulling all data into Spark.

Supported push down interfaces include:

• SupportsPushDownLimit
• SupportsPushDownFilters
• SupportsPushDownAggregates
• SupportsPushDownRequiredColumns

To achieve high performance when writing to ClickHouse, the connector requires the Spark DataFrame to be organized to match the ClickHouse data layout (LSM-like format). This minimizes the need for frequent random insertions and favors batch appends.

Depending on the table type, the connector enforces specific sorting and clustering requirements:

• Distributed tables: The DataFrame must be clustered by [sharding keys, partition keys] and sorted by [sharding keys, partition keys, ordering keys].
• Standard *MergeTree tables: The DataFrame must be sorted by [partition keys, ordering keys].

Note on Spark Limitations: Spark does not support expressions in sharding keys and partition keys without [SPARK-39607].

The project uses Testcontainers and Docker Compose for integration testing. You must have Docker and Docker Compose installed on your system before running tests.

Common Test Commands

Run all tests:

./gradlew clean test

Run all tests with specific Spark and Scala versions (Spark 3.3 and Scala 2.13):

./gradlew clean test -Dspark_binary_version=3.3 -Dscala_binary_version=2.13

Run a single specific test:

./gradlew test --tests=ConvertDistToLocalWriteSuite

Run tests against a custom ClickHouse image:

CLICKHOUSE_IMAGE=custom-org/clickhouse-server:custom-tag ./gradlew test

To publish to a private Nexus Repository, you must configure your local Gradle environment with the necessary credentials and repository URLs. Add the following properties to your ~/.gradle/gradle.properties file:

mavenUser=xxx
mavenPassword=xxx
mavenReleasesRepo=xxx
mavenSnapshotsRepo=xxx

To publish the project to the configured Maven Repository (such as a private Nexus instance), run the Gradle publish task from the root directory.

./gradlew publish

To use the connector in a Gradle project, add the clickhouse-spark-runtime dependency and the clickhouse-jdbc dependency with the all classifier. It is important to set transitive = false for the JDBC dependency to avoid conflicts.

If you are using a SNAPSHOT version, you must also add the Sonatype OSS Snapshots repository to your repositories block.

dependencies {
  implementation("com.clickhouse.spark:clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}")
  implementation("com.clickhouse:clickhouse-jdbc:{{ clickhouse_jdbc_version }}:all") { transitive = false }
}

// Required for SNAPSHOT versions
repositories {
  maven { url = "https://oss.sonatype.org/content/repositories/snapshots" }
}

To use the Spark ClickHouse Connector, ensure the following prerequisites are met:

1. Knowledge: Basic understanding of Apache Spark and ClickHouse.
2. ClickHouse Instance: An available ClickHouse single node or cluster.
3. Spark Cluster: An available Spark cluster compatible with the connector version (see Compatible Matrix).
4. Network Connectivity: Both Spark driver and executor must have network access to the ClickHouse HTTP port. If connecting to a ClickHouse cluster, ensure connectivity between the Spark driver/executors and every node in the ClickHouse cluster.

Once your Python environment is set up, install the necessary dependencies for the documentation site using pip.

pip install -r requirements.txt

To stop the playground and clean up resources:

1. Stop the compose services by pressing CTRL+C in the terminal where they are running.
2. Remove the stopped containers: docker compose --file docker/compose.yml rm

To avoid bundling the connector with every Spark application or using the --jar flag in spark-shell or spark-sql, you can install the required JAR files directly into your Spark installation.

Place the following two files into $SPARK_HOME/jars/:

1. clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}-{{ stable_version }}.jar
2. clickhouse-jdbc-{{ clickhouse_jdbc_version }}-all.jar

Note: For production SQL-only use cases, Apache Kyuubi is recommended.