Omar Atie created CAMEL-24245:
---------------------------------
Summary: Camel-ClickHouse New Component Proposal
Key: CAMEL-24245
URL: https://issues.apache.org/jira/browse/CAMEL-24245
Project: Camel
Issue Type: New Feature
Reporter: Omar Atie
Assignee: Omar Atie
I'd like to propose a new component for integrating with {_}ClickHouse{_}, the
high-performance columnar OLAP database.
Camel can talk to ClickHouse today through the generic xref camel-jdbc /
camel-sql components, but only over the JDBC PreparedStatement path. That works
for low-volume CRUD, but it leaves ClickHouse's high-throughput ingestion
features on the table: native RowBinary/format streaming inserts, server-side
asynchronous inserts, bulk load from files, and compression. Users building
analytics and observability pipelines currently hand-roll beans around the
ClickHouse client to get acceptable ingest performance.
The idea is a camel-clickhouse component built on the official ClickHouse Java
client (client-v2, com.clickhouse, available in Maven Central — the same
library that backs the ClickHouse JDBC driver) that would expose ClickHouse's
native capabilities as first-class endpoint options.
clickhouse://my_db.events?operation=insert&format=RowBinary&batchSize=5000
This follows the pattern already used by camel-influxdb2 (a dedicated component
on a vendor client, rather than generic JDBC), which is the closest analogue in
the catalog.
h2. _Why a dedicated component (vs camel-jdbc)_
- _Native batch insert_ via \{{Client.insert(table, List<?>, InsertSettings)}}
and RowBinary — significantly faster than JDBC \{{addBatch()/executeBatch()}}
for large volumes.
- _Asynchronous inserts_ (\{{async_insert=1}}) for high-concurrency,
small-batch ingestion without client-side buffering.
- _Format streaming_ — stream JSONEachRow / CSV / TSV / Parquet bodies straight
to the server with no per-row serialization.
- _Bulk load from files_ (\{{INSERT ... FROM INFILE}}) with compression
(lz4/zstd).
- _Idiomatic options_ — database, table, format, batchSize, compression, async
— instead of opaque JDBC URL params.
h2. _Design_
- _Producer-only_ (like camel-jdbc): ClickHouse is ingest-via-producer; OLAP
querying is request/reply.
- _Operations:_ \{{insert}} (default), \{{query}}, \{{ping}}.
- _Body types accepted for insert:_ \{{List<Map<String,Object>>}},
\{{List<POJO>}}, JSON/CSV/TSV String or InputStream (matched to \{{format}}),
or a \{{java.io.File}} for bulk load.
- _Client sharing:_ autowire a shared \{{com.clickhouse.client.api.Client}}
bean, or configure \{{serverUrl}}/\{{username}}/\{{password}} on the endpoint.
- _Tests:_ ClickHouse Testcontainers via a new \{{camel-test-infra-clickhouse}}
module; AssertJ assertions.
h2. _Use Cases_
{{_}}Use Case 1: High-throughput event ingestion from Kafka{{_}}
Stream events from Kafka and batch-insert them into ClickHouse using the native
RowBinary format for maximum ingest performance.
{code:java}
from("kafka:events?groupId=analytics")
.aggregate(constant(true), new GroupedBodyAggregationStrategy())
.completionSize(5000).completionTimeout(2000)
.to("clickhouse://analytics.events?operation=insert&format=RowBinary&batchSize=5000")
.log("Inserted ${header.CamelClickHouseWrittenRows} rows");
{code}
{{_}}Use Case 2: Server-side asynchronous inserts for many small producers{{_}}
Let ClickHouse buffer and flush inserts server-side, ideal for many concurrent
producers sending small payloads.
{code:java}
from("platform-http:/ingest")
.unmarshal().json()
.to("clickhouse://metrics.samples?operation=insert&asyncInsert=true&waitForAsyncInsert=false");
{code}
{{_}}Use Case 3: Scheduled OLAP query feeding a dashboard/alert{{_}}
Run an aggregation query on a timer and route the result set to downstream
systems.
{code:java}
from("timer:rollup?period=60000")
.setBody(constant(
"SELECT toStartOfMinute(ts) AS minute, count() AS hits " +
"FROM [analytics.events|http://analytics.events/] WHERE ts > now() - INTERVAL 5
MINUTE " +
"GROUP BY minute ORDER BY minute"))
.to("clickhouse://analytics?operation=query&format=JSONEachRow")
.to("kafka:rollup-metrics");
{code}
{{_}}Use Case 4: Bulk load from files (CSV/Parquet) with compression{{_}}
Ingest data files dropped into a directory using ClickHouse's native file load
with zstd compression.
{code:java}
from("file:data/incoming?include=.*\\.csv.zst&move=.done")
.to("clickhouse://warehouse.orders?operation=insert&format=CSV&compression=zstd")
.log("Loaded file ${header.CamelFileName} into ClickHouse");
{code}
{{_}}Use Case 5: ETL — migrate/aggregate from OLTP into ClickHouse{{_}}
Read rows from a relational source and continuously roll them into ClickHouse
for analytics, decoupling reporting load from the OLTP database.
{code:java}
from("sql:SELECT * FROM orders WHERE exported = false?dataSource=#pg")
.split(body()).streaming()
.aggregate(constant(true), new GroupedBodyAggregationStrategy())
.completionSize(10000).completionTimeout(5000)
.to("clickhouse://warehouse.orders_fact?operation=insert&format=JSONEachRow");
{code}
{{_}}Use Case 6: Observability — write application/access logs to
ClickHouse{{_}}
Fan structured log events into ClickHouse as a cost-effective, queryable log
store.
{code:java}
from("direct:appLog")
.marshal().json()
.to("clickhouse://logs.app_logs?operation=insert&format=JSONEachRow&asyncInsert=true");
{code}
{{_}}Use Case 7: Health check / readiness probe{{_}}
Verify connectivity to the ClickHouse cluster before a route starts processing.
{code:java}
from("timer:health?period=30000")
.to("clickhouse://default?operation=ping")
.choice()
.when(header("CamelClickHousePingOk").isEqualTo(true))
.to("direct:markHealthy")
.otherwise()
.to("direct:alertOps")
.end();
{code}
h2. _Proposed URI options (initial)_
- \{{serverUrl}} — ClickHouse HTTP endpoint (e.g.
[http://localhost:8123|http://localhost:8123/]), or autowire a shared Client
bean
- \{{database}} / table via path — \{{clickhouse://<database>.<table>}}
- \{{operation}} — insert | query | ping (default: insert)
- \{{format}} — RowBinary | JSONEachRow | CSV | TSV | Parquet ... (default:
JSONEachRow)
- \{{batchSize}} — client-side batch size for insert
- \{{asyncInsert}} / \{{waitForAsyncInsert}} — server-side async insert
- \{{compression}} — none | lz4 | zstd
- \{{username}} / \{{password}} (secret) / \{{ssl}}
h2. _Proposed message headers_
- \{{CamelClickHouseOperation}} — override the endpoint operation
- \{{CamelClickHouseDatabase}} / \{{CamelClickHouseTable}} — override target
- \{{CamelClickHouseFormat}} — override format
- \{{CamelClickHouseWrittenRows}} — (out) rows written on insert
- \{{CamelClickHousePingOk}} — (out) boolean result of a ping
I'm happy to implement this and follow the camel-influxdb2 layout, add a
camel-test-infra-clickhouse module with Testcontainers, and provide docs + an
upgrade-guide entry. Feedback on the operation set and default format is
welcome.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
