[
https://issues.apache.org/jira/browse/CAMEL-24245?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18098334#comment-18098334
]
Omar Atie commented on CAMEL-24245:
-----------------------------------
╔══════════════════════════════════════════════════════════════╗
║ camel-clickhouse use-case demo (live ClickHouse validation) ║
╚══════════════════════════════════════════════════════════════╝
Server: http://localhost:8123
Target: camel_demo.events
Database camel_demo.events ready (truncated).
=== 1. PING — health check ===
Camel route returned ping OK
PASS
=== 2. INSERT String (JSONEachRow) ===
DB validated: row from String insert (id=1 AND name='alice' AND
source='string')
PASS
=== 3. INSERT byte[] (JSONEachRow) ===
DB validated: row from byte[] insert (id=2 AND name='bob' AND source='bytes')
PASS
=== 4. INSERT File (JSONEachRow) ===
DB validated: row from File insert (id=3 AND name='carol' AND source='file')
PASS
=== 5. INSERT CSV format ===
DB validated: row from CSV insert (id=4 AND name='dave' AND source='csv')
PASS
=== 6. QUERY — count rows ===
Query returned count=4, readRows=1
PASS
=== 7. QUERY — select rows (CSV) ===
Query result:
1,"alice"
PASS
=== 8. asyncInsert=true ===
DB validated: async insert row (id=5 AND name='eve')
CamelClickHouseWrittenRows header: 0
PASS
=== 9. Header overrides (FORMAT + TABLE) ===
DB validated: row via header table/format (id=6 AND name='frank' AND
source='headers')
PASS
=== 10. batchSize — List body split into batches ===
DB validated: batch row a (id=10 AND name='batch-a')
DB validated: batch row e (id=14 AND name='batch-e')
batch source rows in DB: 5
PASS
=== 11. Default operation (insert) ===
DB validated: default operation insert (id=7 AND name='grace' AND
source='default-op')
PASS
══════════════════════════════════════════════════════════════
Results: 11 passed, 0 failed
Total rows in camel_demo.events: 12
══════════════════════════════════════════════════════════════
┌─id─┬─name────┬─source─────┐
1. │ 1 │ alice │ string │
2. │ 2 │ bob │ bytes │
3. │ 3 │ carol │ file │
4. │ 4 │ dave │ csv │
5. │ 5 │ eve │ async │
6. │ 6 │ frank │ headers │
7. │ 7 │ grace │ default-op │
8. │ 10 │ batch-a │ batch │
9. │ 11 │ batch-b │ batch │
10. │ 12 │ batch-c │ batch │
11. │ 13 │ batch-d │ batch │
12. │ 14 │ batch-e │ batch │
└────┴─────────┴────────────┘
┌─source─────┬─rows─┐
1. │ async │ 1 │
2. │ batch │ 5 │ ← batchSize=2, 5 POJOs → 3 inserts
3. │ bytes │ 1 │
4. │ csv │ 1 │
5. │ default-op │ 1 │
6. │ file │ 1 │
7. │ headers │ 1 │
8. │ string │ 1 │
└────────────┴──────┘
The direct DB validation query completed successfully. It confirmed **12 rows**
in `camel_demo.events`, with one row per source (`string`, `bytes`, `file`,
`csv`, `async`, `headers`, `default-op`) and **5 batch rows** from the
`batchSize=2` List insert test.
> 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)
