[
https://issues.apache.org/jira/browse/CAMEL-24245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Omar Atie updated CAMEL-24245:
------------------------------
Description:
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), 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.
was:
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.
> 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
> Priority: Major
>
> 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), 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)
