apache · GabrielBBaldez · Jun 29, 2026
diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.clickhouse.md b/docs/docs-cn/source/5.application-development/3.connector/11.clickhouse.md
@@ -0,0 +1,67 @@
+# ClickHouse Connector介绍
+ClickHouse Connector支持读和写。
+
+写入时会先将数据缓冲，再按批量（batch）写入ClickHouse。由于ClickHouse是列式存储、针对批量写入做了优化，批量写入相比逐行写入（通过通用JDBC Connector）有显著更高的吞吐。批次大小可配置。
+
+读取时可以按某个数值列将读取拆分为多个分区，从而并行读取。
+
+## 语法
+
+```sql
+CREATE TABLE clickhouse_table (
+  id BIGINT,
+  name VARCHAR,
+  score DOUBLE
+) WITH (
+	type='clickhouse',
+    geaflow.dsl.clickhouse.url = 'jdbc:clickhouse://localhost:8123/default',
+    geaflow.dsl.clickhouse.username = 'default',
+    geaflow.dsl.clickhouse.password = '',
+    geaflow.dsl.clickhouse.table.name = 'source_table'
+);
+```
+
+## 参数
+
+| 参数名 | 是否必须 | 描述                                                |
+| -------- |------|---------------------------------------------------|
+| geaflow.dsl.clickhouse.url     | 是    | The ClickHouse JDBC url, e.g. jdbc:clickhouse://host:8123/database. |
+| geaflow.dsl.clickhouse.table.name     | 是    | The ClickHouse table name. |
+| geaflow.dsl.clickhouse.username     | 否    | The ClickHouse username, default 'default'. |
+| geaflow.dsl.clickhouse.password     | 否    | The ClickHouse password, default empty. |
+| geaflow.dsl.clickhouse.write.batch.size     | 否    | The number of rows buffered before a batch is flushed to ClickHouse. Larger batches give higher write throughput. Default 1000. |
+| geaflow.dsl.clickhouse.driver     | 否    | The ClickHouse JDBC driver class, default com.clickhouse.jdbc.ClickHouseDriver. |
+| geaflow.dsl.clickhouse.partition.num     | 否    | The number of source read partitions, default 1. |
+| geaflow.dsl.clickhouse.partition.column     | 否    | The column used to split the source read into partitions. Default value is 'id'. |
+| geaflow.dsl.clickhouse.partition.lowerbound     | 否    | The lowerbound used to decide the partition stride, not for filtering the rows in the table. |
+| geaflow.dsl.clickhouse.partition.upperbound     | 否    | The upperbound used to decide the partition stride, not for filtering the rows in the table. |
+
+## 示例
+
+```sql
+set geaflow.dsl.clickhouse.url = 'jdbc:clickhouse://localhost:8123/default';
+set geaflow.dsl.clickhouse.username = 'default';
+set geaflow.dsl.clickhouse.password = '';
+
+CREATE TABLE clickhouse_source_table (
+  id BIGINT,
+  name VARCHAR,
+  score DOUBLE
+) WITH (
+	type='clickhouse',
+    geaflow.dsl.clickhouse.table.name = 'source_table'
+);
+
+CREATE TABLE clickhouse_sink_table (
+  id BIGINT,
+  name VARCHAR,
+  score DOUBLE
+) WITH (
+	type='clickhouse',
+    geaflow.dsl.clickhouse.table.name = 'sink_table',
+    geaflow.dsl.clickhouse.write.batch.size = '2000'
+);
+
+INSERT INTO clickhouse_sink_table
+SELECT * FROM clickhouse_source_table;
+```
diff --git a/docs/docs-cn/source/5.application-development/3.connector/index.rst b/docs/docs-cn/source/5.application-development/3.connector/index.rst
@@ -16,4 +16,5 @@
    8.hudi.md
    9.pulsar.md
    10.udc.md
+   11.clickhouse.md
 
diff --git a/docs/docs-en/source/5.application-development/3.connector/11.clickhouse.md b/docs/docs-en/source/5.application-development/3.connector/11.clickhouse.md
@@ -0,0 +1,67 @@
+# ClickHouse Connector Introduction
+The ClickHouse Connector supports both reading and writing operations.
+
+The sink buffers rows and flushes them to ClickHouse in bulk (batched inserts). Because ClickHouse is columnar and optimized for bulk ingestion, batched writes give far higher throughput than inserting row by row through the generic JDBC connector. The batch size is configurable.
+
+The source can split a read into several partitions over a numeric column so the partitions can be read in parallel.
+
+## Syntax
+
+```sql
+CREATE TABLE clickhouse_table (
+  id BIGINT,
+  name VARCHAR,
+  score DOUBLE
+) WITH (
+	type='clickhouse',
+    geaflow.dsl.clickhouse.url = 'jdbc:clickhouse://localhost:8123/default',
+    geaflow.dsl.clickhouse.username = 'default',
+    geaflow.dsl.clickhouse.password = '',
+    geaflow.dsl.clickhouse.table.name = 'source_table'
+);
+```
+
+## Options
+
+| Key | Required | Description |
+| -------- |------|---------------------------------------------------|
+| geaflow.dsl.clickhouse.url     | true    | The ClickHouse JDBC url, e.g. jdbc:clickhouse://host:8123/database. |
+| geaflow.dsl.clickhouse.table.name     | true    | The ClickHouse table name. |
+| geaflow.dsl.clickhouse.username     | false    | The ClickHouse username, default 'default'. |
+| geaflow.dsl.clickhouse.password     | false    | The ClickHouse password, default empty. |
+| geaflow.dsl.clickhouse.write.batch.size     | false    | The number of rows buffered before a batch is flushed to ClickHouse. Larger batches give higher write throughput. Default 1000. |
+| geaflow.dsl.clickhouse.driver     | false    | The ClickHouse JDBC driver class, default com.clickhouse.jdbc.ClickHouseDriver. |
+| geaflow.dsl.clickhouse.partition.num     | false    | The number of source read partitions, default 1. |
+| geaflow.dsl.clickhouse.partition.column     | false    | The column used to split the source read into partitions. Default value is 'id'. |
+| geaflow.dsl.clickhouse.partition.lowerbound     | false    | The lowerbound used to decide the partition stride, not for filtering the rows in the table. |
+| geaflow.dsl.clickhouse.partition.upperbound     | false    | The upperbound used to decide the partition stride, not for filtering the rows in the table. |
+
+## Example
+
+```sql
+set geaflow.dsl.clickhouse.url = 'jdbc:clickhouse://localhost:8123/default';
+set geaflow.dsl.clickhouse.username = 'default';
+set geaflow.dsl.clickhouse.password = '';
+
+CREATE TABLE clickhouse_source_table (
+  id BIGINT,
+  name VARCHAR,
+  score DOUBLE
+) WITH (
+	type='clickhouse',
+    geaflow.dsl.clickhouse.table.name = 'source_table'
+);
+
+CREATE TABLE clickhouse_sink_table (
+  id BIGINT,
+  name VARCHAR,
+  score DOUBLE
+) WITH (
+	type='clickhouse',
+    geaflow.dsl.clickhouse.table.name = 'sink_table',
+    geaflow.dsl.clickhouse.write.batch.size = '2000'
+);
+
+INSERT INTO clickhouse_sink_table
+SELECT * FROM clickhouse_source_table;
+```
diff --git a/docs/docs-en/source/5.application-development/3.connector/index.rst b/docs/docs-en/source/5.application-development/3.connector/index.rst
@@ -15,4 +15,5 @@ Connector
    7.hbase.md
    8.hudi.md
    9.pulsar.md
-   10.udc.md
+   10.udc.md
+   11.clickhouse.md
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/pom.xml
@@ -0,0 +1,75 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you under the Apache License, Version 2.0 (the
+  ~ "License"); you may not use this file except in compliance
+  ~ with the License.  You may obtain a copy of the License at
+  ~
+  ~   http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing,
+  ~ software distributed under the License is distributed on an
+  ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  ~ KIND, either express or implied.  See the License for the
+  ~ specific language governing permissions and limitations
+  ~ under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.geaflow</groupId>
+        <artifactId>geaflow-dsl-connector</artifactId>
+        <version>0.8.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>geaflow-dsl-connector-clickhouse</artifactId>
+    <name>geaflow-dsl-connector-clickhouse</name>
+
+    <properties>
+        <clickhouse-jdbc.version>0.6.5</clickhouse-jdbc.version>
+        <httpclient5.version>5.2.1</httpclient5.version>
+        <testcontainers.version>1.19.8</testcontainers.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-connector-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>${clickhouse-jdbc.version}</version>
+        </dependency>
+        <!-- ClickHouse JDBC talks to the server over HTTP using Apache HttpClient5. -->
+        <dependency>
+            <groupId>org.apache.httpcomponents.client5</groupId>
+            <artifactId>httpclient5</artifactId>
+            <version>${httpclient5.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>${testng.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>
diff --git a/...house/src/main/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseConfigKeys.java b/...house/src/main/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseConfigKeys.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.clickhouse;
+
+import org.apache.geaflow.common.config.ConfigKey;
+import org.apache.geaflow.common.config.ConfigKeys;
+
+public class ClickHouseConfigKeys {
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_DRIVER = ConfigKeys
+        .key("geaflow.dsl.clickhouse.driver")
+        .defaultValue("com.clickhouse.jdbc.ClickHouseDriver")
+        .description("The ClickHouse JDBC driver class, "
+            + "default com.clickhouse.jdbc.ClickHouseDriver.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_URL = ConfigKeys
+        .key("geaflow.dsl.clickhouse.url")
+        .noDefaultValue()
+        .description("The ClickHouse JDBC url, e.g. jdbc:clickhouse://host:8123/database.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_USERNAME = ConfigKeys
+        .key("geaflow.dsl.clickhouse.username")
+        .defaultValue("default")
+        .description("The ClickHouse username, default 'default'.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_PASSWORD = ConfigKeys
+        .key("geaflow.dsl.clickhouse.password")
+        .defaultValue("")
+        .description("The ClickHouse password, default empty.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_TABLE_NAME = ConfigKeys
+        .key("geaflow.dsl.clickhouse.table.name")
+        .noDefaultValue()
+        .description("The ClickHouse table name.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_WRITE_BATCH_SIZE = ConfigKeys
+        .key("geaflow.dsl.clickhouse.write.batch.size")
+        .defaultValue(1000)
+        .description("The number of rows buffered before a batch is flushed to ClickHouse. "
+            + "ClickHouse is columnar and strongly prefers bulk inserts, so larger batches "
+            + "give much higher write throughput. Default 1000.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_PARTITION_NUM = ConfigKeys
+        .key("geaflow.dsl.clickhouse.partition.num")
+        .defaultValue(1L)
+        .description("The number of source read partitions, default 1.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_PARTITION_COLUMN = ConfigKeys
+        .key("geaflow.dsl.clickhouse.partition.column")
+        .defaultValue("id")
+        .description("The column used to split the source read into partitions.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_PARTITION_LOWERBOUND = ConfigKeys
+        .key("geaflow.dsl.clickhouse.partition.lowerbound")
+        .defaultValue(0L)
+        .description("The lowerbound used to decide the partition stride, "
+            + "not for filtering the rows in the table.");
+
+    public static final ConfigKey GEAFLOW_DSL_CLICKHOUSE_PARTITION_UPPERBOUND = ConfigKeys
+        .key("geaflow.dsl.clickhouse.partition.upperbound")
+        .defaultValue(0L)
+        .description("The upperbound used to decide the partition stride, "
+            + "not for filtering the rows in the table.");
+}
diff --git a/...e/src/main/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnector.java b/...e/src/main/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnector.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.clickhouse;
+
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.dsl.connector.api.TableReadableConnector;
+import org.apache.geaflow.dsl.connector.api.TableSink;
+import org.apache.geaflow.dsl.connector.api.TableSource;
+import org.apache.geaflow.dsl.connector.api.TableWritableConnector;
+
+public class ClickHouseTableConnector implements TableReadableConnector, TableWritableConnector {
+
+    public static final String TYPE = "CLICKHOUSE";
+
+    @Override
+    public String getType() {
+        return TYPE;
+    }
+
+    @Override
+    public TableSource createSource(Configuration conf) {
+        return new ClickHouseTableSource();
+    }
+
+    @Override
+    public TableSink createSink(Configuration conf) {
+        return new ClickHouseTableSink();
+    }
+}