[
https://issues.apache.org/jira/browse/CAMEL-24245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Omar Atie updated CAMEL-24245:
------------------------------
Labels: new-feature (was: )
> 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
> Labels: new-feature
> Attachments: camel-clickhouse-demo.zip
>
>
> 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 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)
