DEV Community: Shahab Ranjbary

Immutable Data ClickHouse, Mutable Data StarRocks: Why I Moved My CDC Tables

Shahab Ranjbary — Fri, 09 Oct 2026 11:32:04 +0000

TL;DR — ClickHouse is still the tool I reach for when data is written once and never changes: events, logs, clicks. But a big part of our analytics data is a copy of OLTP tables, and those rows get updated and deleted all the time. For that job I moved to StarRocks Primary Key tables: upserts and deletes are applied at write time, and a plain SELECT returns the current row. This post covers why, the trade-offs, and the honest list of things that went wrong.

The problem: mutable data

Some data never changes after it's written: a click, a log line, a page view. You append it and you aggregate it.

Other data changes all the time. An order is created, paid, packed, delivered, and sometimes cancelled or deleted. When you replicate tables like that from MySQL with CDC (change data capture), you get a stream of inserts, updates and deletes, and the analytics copy must always show the current row.

These two kinds of data need different things from a database. This post is about the second kind.

Why ClickHouse was painful for this job

Let me be fair first: ClickHouse is excellent. It's fast, stable and mature, and for append-only data I still think it's the best choice. The friction came from asking it to do something it wasn't designed for: keeping a mutable copy of an OLTP database.

1. "Current state" is eventual, not immediate

The usual CDC pattern in ClickHouse is ReplacingMergeTree(version). Every update is a new row, and duplicates are removed when parts merge, at some unknown time later. Until then the table holds several versions of the same key.

So every query has to choose:

-- Option A: FINAL (correct, but it does merge work on every read)
SELECT status, count() FROM orders FINAL GROUP BY status;

-- Option B: argMax per key (correct, but every query gets more complex)
SELECT status, count() FROM (
  SELECT id, argMax(status, _version) AS status,
             argMax(is_deleted, _version) AS is_deleted
  FROM orders GROUP BY id
) WHERE is_deleted = 0
GROUP BY status;

Forget FINAL in one dashboard query and you double-count orders. Analysts will forget.

2. Deletes are a flag, not a delete

A CDC delete becomes a tombstone row (is_deleted = 1) that every query must filter out. Newer versions help (ReplacingMergeTree(ver, is_deleted), lightweight deletes), but the read-side discipline is still there.

3. Joins across many mutable tables

Reports join many of these tables: orders × order items × payments × customers... When every side of the join needs FINAL or an argMax subquery, the SQL gets heavy and the planner has less room to optimize.

4. Three objects per table

Kafka engine table → materialized view → target table. Multiply by a few hundred tables and schema changes become painful.

5. Small Avro surprises

Two I hit on ClickHouse 25.3:

A Debezium topic.prefix with a hyphen (like my-app) produces an Avro namespace that ClickHouse's Avro reader rejects (Invalid namespace). Fix: schema.name.adjustment.mode: avro on the connector, or just no hyphens.
An Avro string (Debezium decimal.handling.mode=string) can't be read straight into a Decimal column. The Kafka table needs Nullable(String), and the MV casts it.

None of these is a dealbreaker on its own. Together, they meant that every consumer of the data had to know it came from CDC. I wanted a database where SELECT * FROM orders WHERE id = 42 just returns the current row.

Why StarRocks fits mutable data

Primary Key tables: upsert and delete at write time

CREATE TABLE orders (
    id          BIGINT        NOT NULL,
    created_at  DATETIME      NOT NULL,
    status      VARCHAR(32)   NULL,
    total_price DECIMAL(18,2) NULL,
    updated_at  DATETIME      NULL,
    _version    BIGINT        NOT NULL DEFAULT "0"
)
PRIMARY KEY (id, created_at)
PARTITION BY date_trunc('month', created_at)
DISTRIBUTED BY HASH(id);

A Primary Key table does delete-then-insert when the data is written. No merge-on-read and no FINAL. A reader always sees exactly one row per key: the latest one.

Deletes use a special column, __op (0 = upsert, 1 = delete), so a CDC delete really removes the row. With a Kafka load it's one line in the column mapping:

CREATE ROUTINE LOAD shop.load_orders ON orders
COLUMNS (
    id, created_at, status, total_price, updated_at, op, ts_ms,
    _version = ts_ms,
    __op     = if(op = 'd', 1, 0)        -- the important line
)
PROPERTIES ("format" = "avro")
FROM KAFKA (
    "kafka_topic" = "mysql.shop.orders",
    "confluent.schema.registry.url" = "http://schema-registry:8081"
);

Ordering comes from Kafka: Debezium keys messages by primary key, so all events for one row land in one partition, in order.

Conditional updates for back-fills

When you back-fill history from a snapshot while the live stream keeps running, an old snapshot row must not overwrite a newer live row. StarRocks handles this with merge_condition:

PROPERTIES ("merge_condition" = "_version")
-- the incoming row is applied only if its _version >= the stored one

It speaks MySQL

StarRocks uses the MySQL wire protocol and a very similar SQL dialect. Reconciliation queries run unchanged on the source and on the copy, and analysts connect with the same DBeaver/Metabase drivers. If you check a replica against MySQL often, that matters more than it sounds.

Joins are a first-class citizen

A cost-based optimizer, shuffle/broadcast/colocate joins, and async materialized views. Tables that share a key (orders and order_items on order_id) can be colocated, so the join never moves data across the network.

Partial updates and schema changes

Partial column updates (fill one column without rewriting the row) saved me several times when I had to repair one column across millions of rows. fast_schema_evolution makes ADD COLUMN nearly instant.

The cons (the honest part)

StarRocks is not magic. Here's what bit me or what you should plan for.

1. Silent NULLs with `strict_mode = false`

With strict mode off, a value that doesn't fit its column becomes NULL instead of failing the load:

VARCHAR(n) in StarRocks counts bytes, MySQL counts characters. Persian text is 2 bytes per character, so long values became NULL with no error. Size VARCHARs as MySQL length × 4.
A MySQL ENUM('NO','EXTERNAL',...) mapped to BOOLEAN → 100% NULL. Zero errors.

Lesson: after loading a table, compare NULL rates per column with the source. "Loaded rows" and "0 error rows" don't prove the values are right.

2. Your primary key must never change at the source

In a Primary Key table the partition column has to be part of the key, and you may add a parent id to the key so related tables can be colocated. If one of those columns changes in MySQL, the update arrives with a new key: StarRocks inserts a new row and the old one stays forever. I found thousands of such "ghost rows" on one table after a bulk change of a parent id.

Lesson: build the PK only from columns that never change (id, created_at). If you can't, plan a regular check for keys that exist in StarRocks but not in MySQL.

3. `merge_condition` can't protect against a delete

Conditional updates compare the incoming row with the stored one, and they don't work together with deletes, so use them for back-fills only. Side effect: if MySQL deletes a row while a back-fill snapshot is running, the snapshot can re-insert it later, because there is no stored row left to compare against. It's rare, but real. Compare keys with the source after every back-fill.

4. The primary key index costs memory

Fast upserts come from a primary key index. Keep keys small (integers, not long strings), and turn on the persistent (on-disk) index for big tables. A wide VARCHAR key on a billion-row table gets expensive.

5. Throughput scales with backends

Every upsert does a key lookup, and compaction runs after big loads. On a small cluster, back-fills of hundreds of millions of rows take hours, and running them in parallel doesn't help much. Slice them by primary-key range and plan the time.

6. Smaller ecosystem

ClickHouse has a bigger community, more integrations, more Stack Overflow answers and more managed offerings. With StarRocks I read source code more than once to understand a behavior.

7. Now there are two databases to run

Keeping ClickHouse for event data means two systems to operate, monitor and upgrade. For us the clean split was worth it, but it is a real cost.

Side by side

	ClickHouse	StarRocks
Best at	Append-only, immutable data	Mutable data, upserts, CDC
Updates	New version row, deduplicated at merge time	Delete+insert at write time (PK table)
Deletes	Tombstone flag or mutation	Real delete via `__op`
Read latest state	`FINAL` / `argMax`	Plain `SELECT`
Joins	Improving fast; best denormalized	CBO, colocate/shuffle joins
Protocol	Native + HTTP	MySQL wire protocol
Kafka ingest	Kafka engine + MV (per table)	Routine Load or Stream Load
Out-of-order guard	Version column	`merge_condition`
Ecosystem	Large, mature	Smaller, growing
Watch out for	Forgotten `FINAL`, merge lag	Silent NULLs, PK design, PK index memory

Where we ended up

ClickHouse stays for immutable, high-volume event data. It's still the best tool for that.
StarRocks holds the mutable tables we replicate from MySQL.
Analysts query current state with plain SQL. No FINAL, no is_deleted filters, no "remember this table comes from CDC".

The image at the top says it better than I can: two bottles, two openers. Don't force one database to do both jobs.

Checklist if you're doing the same

Pick the engine by how the data changes, not by benchmark charts.
Build primary keys only from columns that never change.
Test deletes end to end before going live, not just inserts.
Check NULL rates and values against the source after loading. "0 error rows" is not "correct".
Use merge_condition for back-fills, and check for re-inserted deleted rows afterwards.
Monitor freshness (newest update time per table), not just load status.

Have you run CDC into ClickHouse or StarRocks? I'd love to hear what bit you. Let me know in the comments. 👇

From SSH Chaos to Self-Service GPU Workspaces with Coder and HAMi

Shahab Ranjbary — Thu, 09 Jul 2026 21:32:52 +0000

Disclosure: I designed the architecture, production rollout, and validation for this project. Parts of the Coder implementation were developed with Claude as an AI coding assistant, while all design decisions, testing, and production validation were performed on a real Kubernetes cluster.

TL;DR: Shared bare-metal GPU box → Coder on Kubernetes with per-user CPU/RAM/disk/VRAM quotas (forked) → HAMi for VRAM-isolated GPU sharing across 3× L40S. Fork: ranjbaryshahab/coder feat/per-user-resource-limits. Real gotchas inside.

The problem: one GPU box, many data scientists, endless conflicts

Our data-science team shared a single bare-metal GPU server over SSH. You can guess how that went: conflicting Python environments, someone's training job eating all the VRAM, "who installed CUDA 12.4 and broke everything," and no isolation between people. It worked — until it didn't, constantly.

So I joined that server to our Kubernetes cluster and started looking for something better.

The false starts (that taught me what I actually needed)

My first instinct was the obvious one: "just use containers." I told the team to work in Docker.

They pushed back — and they were right. For model training they need a real, persistent environment: SSH in, install things, run long jobs, use the GPU directly. A throwaway docker run isn't that.

So I tried the middle ground: one Ubuntu pod on Kubernetes with SSH access for everyone. Better than bare metal… but I'd just recreated the original problem inside a pod. Still no isolation, still no resource limits — one person could still starve everyone else, and I had no idea who was using what.

That's when the real requirements clicked:

Per-user isolation (their own environment, their own home)
Resource quotas (CPU / RAM / disk / GPU VRAM) so nobody hogs the box
Self-service — spin up a workspace, pick VS Code / Cursor / a web IDE, without me in the loop
Still real SSH and GPU for training

Enter Coder

Coder fit almost perfectly: self-hosted developer workspaces on your own infrastructure, with SSH, VS Code Desktop, browser IDEs, and Terraform-defined templates. On Kubernetes each workspace is a pod with a persistent home volume.

Almost — because the one thing I needed most, per-user resource quotas, is an Enterprise feature. Coder OSS doesn't provide pooled per-user resource quotas like "this user gets 32 CPU cores, 64 GB RAM, and 48 GB of VRAM to distribute across all of their workspaces."

So I forked Coder OSS and added the missing capability.

What I added to Coder OSS

A custom_user_limits table + a self-contained API (one Go file, five migrations) — deliberately kept small and additive so the fork stays easy to rebase onto upstream.
Pooled per-user budgets: CPU, memory, disk, and GPU VRAM are a total the user spends across all their workspaces (0 = unlimited).
Enforcement in the Terraform template, not in Coder's core — preconditions reject a build that would blow the budget, with human-readable errors ("CPU quota exceeded: 12 of 32 cores already used…"). Worst case is a failed build with a clear message, never corrupted state.
The template authenticates as the workspace owner (coder_workspace_owner.me.session_token) against self-service endpoints — so there's no admin token baked into templates, nothing to leak or rotate.
Admin Usage page (live per-workspace CPU/mem/disk/VRAM) and a self-service Quota page for users.

One subtle bug worth sharing: the naive pooled check has a race — two workspaces created at the same instant each exclude themselves, count "everyone else," and both slip past the budget. The fix is a deterministic total order (count only workspaces whose build is strictly older), so of two concurrent creates exactly one loses. No locks, no reservation table.

Building a fork with AI assistance

I didn't hand-write every line of the fork. I used Claude to implement much of the Go, SQL, TypeScript, and Terraform after describing the features and constraints in plain language. That sped development considerably, but it didn't replace engineering judgment.

The architecture, trade-offs, production testing, and validation were still mine. In several cases the AI made confident but incorrect assumptions—for example, treating the L40S as having 48 GB of usable VRAM. Production testing exposed those mistakes, and they were corrected before rollout.

AI turned out to be a useful implementation partner, but a poor substitute for verification. Running everything on a real cluster remained essential.

Then the GPUs fought back

With Coder live, I hit the hard problem. We have 3× NVIDIA L40S on that node. Two engineers each created a workspace asking for 48 GB of VRAM — and Kubernetes scheduled them onto the same physical GPU. Neither could train.

That's when I learned NVIDIA time-slicing shares GPU compute time, not memory. The scheduler treats "GPUs" as interchangeable slots and is completely blind to VRAM — so two "48 GB" jobs happily collide on one 46 GB card and OOM each other. The vanilla options all failed my use case:

Time-slicing → no memory isolation (the collision above)
Whole-GPU → only 3 concurrent users, most VRAM wasted
MIG → not supported on L40S
MPS → shares are uniform; can't mix a 48 GB and a 10 GB tenant

I wasn't looking for GPU virtualization—I was looking for VRAM-aware scheduling. That's what eventually led me to HAMi.

HAMi to the rescue

HAMi (a CNCF project) does exactly what I needed: pods request nvidia.com/gpumem in MiB, and HAMi's scheduler bin-packs by free VRAM and hard-caps each pod via CUDA interception. Inside the container, nvidia-smi shows exactly the memory you were granted — nobody can eat a neighbor's VRAM.

Mixed sizes now pack cleanly onto whatever card has room; a request that doesn't fit waits instead of silently colliding.

The gotchas that cost me the most (airgapped cluster + shared prod):

Rename the managed resource so HAMi's webhook never touches our H100 production node — one operator, two GPU nodes, zero blast radius.
Per-node handover via labels — the GPU-Operator keeps the driver/toolkit, HAMi takes scheduling on just the L40S node.
The kube-scheduler image isn't on Docker Hub → needed a mirror.
"48 GB" is marketing; usable is 46,068 MiB. Cap your VRAM picker at 44 GiB or requests never schedule.

Where it landed

The platform is now used daily by our data-science team. Data scientists now open Coder, click New workspace, pick VS Code Desktop / Cursor / a browser IDE, choose their CPU/RAM/VRAM within their budget, and go — SSH and GPU included, fully isolated, no more stepping on each other. I can see exactly who's using what, and one workspace can't starve the box.

And it's all observable — Coder + GPU metrics in Grafana:

Open source

The Coder fork (per-user quotas for OSS) is public:

👉 github.com/ranjbaryshahab/coder — branch feat/per-user-resource-limits

Backend, migrations, the three UI pages, and the race-free quota logic are all there. The GPU enforcement lives in the workspace template (HAMi nvidia.com/gpumem requests) and the deployment repo.

What I'd tell you if you're doing this

Keep a fork minimal and additive. One Go file, isolated migrations, template-side enforcement. Rebasing over upstream stays cheap.
Put enforcement where the blast radius is small. Template preconditions can't corrupt core state — worst case is a clear error.
0 = unlimited needs a fail-closed path, or a broken token silently hands out infinite resources.

What started as a single shared SSH server evolved into a self-service GPU platform with isolated workspaces, pooled user quotas, and VRAM-aware scheduling. More importantly, the operational burden disappeared: data scientists provision their own environments, and the infrastructure enforces fairness automatically.

🧠 ClickHouse LEFT JOINs: Why join_use_nulls Matters

Shahab Ranjbary — Thu, 30 Oct 2025 10:10:28 +0000

🧠 Understanding `join_use_nulls` in ClickHouse

ClickHouse is famous for being blazing fast — but sometimes its SQL semantics can surprise you, especially around JOINs.

Here’s a simple example that shows how the join_use_nulls setting can completely change your results.

🧩 The Setup

Let’s create two tiny tables:

CREATE TABLE test.id_val(`id` UInt32, `val` UInt32) ENGINE = TinyLog;
INSERT INTO test.id_val VALUES (1,11),(2,12),(3,13);

CREATE TABLE test.id_val2(`id` UInt32, `val` UInt32) ENGINE = TinyLog;
INSERT INTO test.id_val2 VALUES (1,21),(1,22),(3,23);

We’ve got:

id_val: three rows with IDs 1, 2, and 3
id_val2: three rows, but ID 2 is missing and ID 1 appears twice

🧮 The Default JOIN

Let’s run a LEFT JOIN:

SELECT *
FROM test.id_val
LEFT JOIN test.id_val2 USING (id);

Output:

┌─id─┬─val─┬─val_1─┐
│  1 │  11 │   21  │
│  1 │  11 │   22  │
│  2 │  12 │    0  │
│  3 │  13 │   23  │
└────┴─────┴───────┘

Notice something interesting?
When there’s no matching row for id=2, ClickHouse fills it with a default value — in this case, 0.

For numeric columns, the default is 0.
For strings, it would be an empty string ''.

🧩 Enter join_use_nulls

If you want standard SQL behavior, where missing values become NULL, use:

SELECT *
FROM test.id_val
LEFT JOIN test.id_val2 USING (id)
SETTINGS join_use_nulls = 1;

Now the output looks like this:

┌─id─┬─val─┬─val_1─┐
│  1 │  11 │   21  │
│  1 │  11 │   22  │
│  2 │  12 │  NULL │
│  3 │  13 │   23  │
└────┴─────┴───────┘

That’s closer to what most SQL engines (like PostgreSQL or MySQL) would do.

⚙️ Why This Matters

This setting affects how ClickHouse represents non-matching rows in joins — which can have downstream effects on:

Aggregations
SELECT count(val_1) FROM ...
will count 0s differently than NULLs.
Exports & BI tools
Some tools treat 0 as a valid number, but NULL as missing data.
Data semantics
A missing match isn’t always “zero” — it might mean “unknown”.

🚀 Performance Note

The default behavior (without join_use_nulls) is slightly faster and more memory-efficient because ClickHouse doesn’t have to track NULL bitmaps.

If you’re doing high-volume analytical joins and don’t care about SQL-standard nulls, keep it off.
If you need correctness and consistency with other systems, turn it on.

🔍 Check Your Settings

You can check the current value of this setting in your ClickHouse session:

SELECT name, value
FROM system.settings
WHERE name LIKE '%join_use_nulls%';

Or set it globally in your configuration file or session:

SET join_use_nulls = 1;

📚 Reference

For more detail on ClickHouse’s Join engine and its settings (including join_use_nulls), see: ClickHouse docs — Join table engine

🗨️ Discussion

Have you ever been surprised by ClickHouse JOIN behavior?
Do you leave join_use_nulls off for performance, or always turn it on for clarity?

🛠️ Copying and Moving Data Between Tables in ClickHouse (with Different Engines)

Shahab Ranjbary — Thu, 17 Apr 2025 00:27:58 +0000

Sometimes in ClickHouse, you need to move or copy data between tables — especially when you're changing the table engine, for example from MergeTree to ReplicatedMergeTree.

At first glance, it may seem like MOVE PARTITION TO TABLE should do the trick — but in many cases, it doesn't work due to strict requirements. Let’s go over why, and a better approach we used.

❌ Why `MOVE PARTITION TO TABLE` Didn’t Work for Us

The ALTER TABLE source MOVE PARTITION TO TABLE dest command physically moves partitions and deletes them from the source table. But it comes with strict requirements:

✅ Same structure
✅ Same partition key
✅ Same primary key
✅ Same order by
✅ Same storage policy
✅ Same engine family (e.g., both must be MergeTree or both ReplicatedMergeTree)
✅ Same or superset of indices and projections

So when you’re trying to move data from a MergeTree table to a ReplicatedMergeTree table, this query will fail because the engines are different.

✅ Alternative: `ATTACH PARTITION FROM`

Instead, we used this command:

ALTER TABLE new_table ATTACH PARTITION partition_expr FROM old_table;

This copies the data (without deleting it from the source) and supports different engine types, which makes it ideal for our use case.

✅ Requirements:

Same structure
Same partition key, order by, and primary key
Same storage policy
Indices/projections must match (or be a superset if enforce_index_structure_match_on_partition_manipulation is disabled)

It’s a safe and clean way to transfer data between tables — even when they use different engines.

🧹 Final Step: Drop the Old Partition

Once the data is successfully attached and validated, we clean up the original table:

ALTER TABLE old_table DROP PARTITION partition_expr;

This removes the partition from the old table (and on all replicas, if any).

Data is physically deleted after ~10 minutes.

🧾 What About `ATTACH TABLE AS REPLICATED`?

ClickHouse also supports converting a table like this:

DETACH TABLE my_table;
ATTACH TABLE my_table AS REPLICATED;
SYSTEM RESTORE REPLICA my_table;

But we chose not to use this option.

Why?

Our goal wasn’t to convert the existing table but to control the replication and migration process manually.

So while ATTACH TABLE AS REPLICATED is valid and useful in some scenarios, it wasn’t the right fit for us.

🧪 Summary: Best Way to Copy Between Tables with Different Engines

🔧 Create a new table with the same schema, keys, and partitioning.
📥 Use ATTACH PARTITION FROM to copy data.
✅ Validate.
🗑️ Drop the original partition if needed.

This approach is clean, safe, and works even when the engines are different.

📚 Resources

🧡 Thanks for reading!
If you’ve used other approaches or faced gotchas while doing engine migrations in ClickHouse, let’s discuss in the comments!

Choosing the Right Real-Time Stream Processing Framework

Shahab Ranjbary — Sun, 08 Sep 2024 23:08:53 +0000

Introduction

In the dynamic realm of real-time stream processing, tools like ksqlDB, Apache Spark, Apache Flink, Apache Storm, Timeplus Proton, and RisingWave each offer distinct advantages tailored to diverse use cases and requirements. This article delves into these leading stream processing frameworks' unique capabilities, use cases, and pros and cons.

ksqlDB

ksqlDB simplifies real-time stream processing by providing a SQL-based layer over Apache Kafka Streams, ensuring seamless integration within Kafka environments.

Use Cases

Materialized Cache: Build and serve incrementally updated stateful views. Useful for applications requiring up-to-date information without continuous database queries.
Streaming ETL Pipeline: Manipulate in-flight data to connect arbitrary sources and sinks. Facilitates seamless real-time data integration across systems.
Event-Driven Microservices: Trigger changes based on observed event patterns in a stream. Ideal for real-time notifications, monitoring, and automated decision-making systems.
Stream Processing Cookbook: Use predefined recipes to quickly implement common streaming data tasks, which benefits beginners and experienced users.

Pros

Strong Kafka Integration: Built-in integration with Apache Kafka for streamlined event-streaming applications.
Low Latency: Delivers latency as low as ten milliseconds, ideal for real-time analytics.
SQL Interface: Familiar SQL interface for easy adoption.
Fault Tolerance: Built-in fault tolerance mechanisms.

Cons

Limited Analytics Capability: Lags behind competitors like Flink and Spark in advanced analytics tools.
Learning Curve: Slightly steeper learning curve compared to simpler frameworks.
ETL Limitations: Not ideal for complex ETL operations. Checkpointing Challenges: Advanced checkpointing mechanisms are lacking, impacting exact-once processing in complex scenarios.

Apache Spark (Structured Streaming)

Apache Spark offers scalability and flexibility through its Structured Streaming component, supporting batch and stream processing.

Use Cases

Streaming ETL: Continuously clean and aggregate data before pushing it into data repositories, simplifying conventional ETL processes.
Data Enrichment: Combine live and historical data for real-time analysis, enhancing data quality for personalized marketing efforts.
Trigger Event Detection: Detect and respond to unusual patterns or events in real-time, such as fraudulent transactions or critical health alerts.
Complex Session Analysis: Analyze user activities in real-time, continuously updating machine learning models. Used by companies like Netflix for real-time show recommendations.

Pros

Language Support: Supports Scala, Java, Python, R, C#, F#.
Community Support: Extensive community and ecosystem support.
Exact-Once Processing: Ensures exactly-once processing semantics.
Versatility: Integrates well with diverse data sources beyond Kafka.
SQL Capabilities: Supports SQL for querying and processing.

Cons

Micro-Batch Processing: Operates on micro-batch processing, potentially limiting ultra-low latency applications.
High Memory Consumption: In-memory processing architecture can lead to high memory usage.
Complexity: Deployment and management complexity, particularly in large-scale environments.
Integration Challenges: Direct SQL querying on Kafka topics is not straightforward.

Apache Flink

Apache Flink excels in low-latency, high-throughput stream processing with unified APIs for both batch and stream processing.

Use Cases

Event-Driven Applications: Real-time fraud detection, anomaly detection, rule-based alerting, business process monitoring, and handling real-time user interactions in social networks.
Data Analytics Applications: Quality monitoring of telco networks, product updates analysis, ad-hoc analysis of live data, and large-scale graph analysis.
Data Pipeline Applications: Real-time search index building and continuous ETL in e-commerce environments.

Pros

Low Latency: Minimal latency and high throughput suitable for demanding real-time applications.
Unified API: Simplifies development with a unified API for batch and stream processing.
Advanced Analytics: Offers advanced analytical functions and robust state management.
Exactly-Once Processing: Supports exactly-once processing semantics.
Windowing Support: Flexible windowing capabilities based on both time and count.

Cons

Integration Complexity: Integration with other big data ecosystems like Hadoop/Hive can be challenging.
Limited Language Support: Supports Java, Scala, Python, and SQL, limiting language flexibility.
Community Size: Smaller community compared to Apache Spark, potentially affecting support availability.
Learning Curve: Requires some learning effort, especially for complex use cases.

Timeplus Proton

Timeplus Proton leverages ClickHouse for efficient, high-speed streaming SQL processing with an Apache 2.0 open-source license.

Use Cases

FinTech: Real-time post-trade analytics and real-time pricing for financial instruments.
DevOps: Real-time observability, metering for usage-based pricing, and real-time GitHub insights.
Security Compliance: SOC2 compliance dashboards and alerts, container vulnerability monitoring, monitoring superblocks user activities, and protecting sensitive information in Slack.
IoT: Real-time fleet monitoring and real-time sensor data from smartphones.
Video Streaming: Analyzing Livepeer video engagement metrics.
Customer 360: Auth0 notifications for new signups, HubSpot custom dashboards/alerts, Jitsu clickstream analysis, and real-time Twitter marketing.
Misc: Wildfire monitoring and alerting, and data-driven parenting tasks.

Pros

High Performance: Utilizes ClickHouse for efficient data processing and storage.
Low Latency: Designed for minimal latency and efficient resource usage.
Flexibility: Supports various data sources beyond Kafka.
Open Source: Apache 2.0 licensed, offering flexibility and community support.

Cons

Limited Kafka Integration: Not deeply integrated with Kafka-specific features compared to other solutions.
Use Case Specific: More suited for specific use cases rather than general-purpose stream processing.

RisingWave

RisingWave, built in Rust, emphasizes efficient I/O handling and lightweight scaling with a custom storage engine for SQL-based stream processing.

Use Cases

Streaming ETL: Continuous data transformation and integration.
Real-Time Analytics: On-the-fly data analysis and reporting.
Event-Driven Applications: Real-time event processing and decision-making.

Pros

Efficient I/O Operations: Optimized for high-performance I/O handling.
LSM-Tree Based Engine: Efficient data management with an LSM-Tree-based storage engine.
SQL Support: Provides SQL-based stream processing with materialized views.
Global Checkpoints: Ensures consistent results with frequent global checkpoints.

Cons

Limited Ecosystem and Community Support: As a newer framework, it may have limited ecosystem integrations and community support.
Scalability Considerations: Scalability challenges may arise as it grows and scales out.

Apache Storm

Apache Storm is an open-source distributed stream processing engine. Mainly written in Java and Clojure, it gained popularity after being acquired by Twitter in 2011. Storm can be considered a pioneer in real-time distributed stream processing.

Use Cases

Real-Time Analytics: Used by companies like Spotify for real-time monitoring and analytics.
Online Machine Learning: Suitable for continuous computation and distributed machine learning tasks.
ETL Processes: Handles complex ETL tasks efficiently.
Distributed RPC: Enables distributed remote procedure calls for real-time applications.

Pros

Real-Time Processing: Designed for low-latency, high-throughput applications.
Flexibility: Can be used with any programming language and integrates well with various data sources.
Scalability: Capable of processing over a million tuples per second per node.
Fault Tolerance: Ensures reliable processing with built-in fault tolerance mechanisms.

Cons

Complexity: Requires careful setup and configuration; can be tricky to install and integrate with other technologies.
Limited Windowing Support: Lacks robust support for event-based time management and windowing.
API Limitations: Does not provide high-level APIs like SQL natively, making it harder for new developers to adopt.
State Management: Stateless by default, which can complicate fault tolerance and recovery mechanisms.

Conclusion

Choosing the right stream processing framework depends on specific requirements such as latency, scalability, integration needs, and operational complexity. Each tool offers unique strengths and trade-offs, making it crucial to align them with project goals and constraints:

ksqlDB excels in Kafka-centric environments but may not meet advanced analytics needs.
Apache Spark (Structured Streaming) provides extensive community support and exact-once processing but operates on micro-batch processing.
Apache Flink is optimal for low-latency, high-analytics environments with a unified API but requires careful integration.
Timeplus Proton offers high performance and flexibility with minimal latency but lacks deep Kafka integration.
RisingWave focuses on efficient I/O handling and SQL-based processing, suitable for specialized applications but with scalability considerations.
Apache Storm provides robust real-time processing capabilities but comes with a steeper learning curve and limited high-level API support.

Additional Insights from Streaming Databases

In today’s data-driven landscape, streaming databases like RisingWave and ksqlDB stand out for their ability to handle high-speed, real-time data ingestion and processing. Unlike stream processing engines focused solely on real-time data manipulation, streaming databases integrate storage capabilities, enabling seamless real-time querying and analytics. This unique combination positions them as pivotal tools for organizations aiming to harness real-time insights and maintain a competitive advantage in a fast-paced digital environment.

References

Streaming Databases: Everything You Wanted to Know
Timeplus vs. ksqlDB
Proton: An Open-Source Alternative to ksqlDB for Streaming Processing
Introducing KSQL: Streaming SQL for Apache Kafka
Choosing a Stream Processing System?
Kafka Streams and ksqlDB Compared – How to Choose
Flink vs. Kafka: A Quick Guide to Stream Processing Engines
Comparing ksqlDB, Spark SQL, and Flink SQL
RisingWave vs ksqlDB
What’s the Difference Between Kafka and Spark?
Why I Recommend My Clients NOT Use KSQL and Kafka Streams Apache Flink vs Apache Storm: Which Tool is Better for Your Next Project?
Kafka Streams vs. Apache Flink vs. Apache Storm: Stream Processing Showdown
Life Happens in Real Time, Not in Batches: Choosing a Data Streaming Platform and Stream Processing Engine

Tracking User Account Changes in Real-Time: A Debezium and ClickHouse Implementation

Shahab Ranjbary — Mon, 15 Jul 2024 18:00:07 +0000

Introduction

In modern e-commerce applications, tracking changes to critical data like user account types in real time is essential for business intelligence and regulatory compliance. Traditional databases often overwrite old values, making historical analysis challenging. Enter Change Data Capture (CDC), a technique that captures and stores every change to your data, enabling a comprehensive audit trail and real-time analytics.

Scenario Overview

Imagine you manage a PostgreSQL database for an e-commerce platform where user account types (Bronze, Silver, Gold) frequently change. However, PostgreSQL only retains the latest values, making it difficult to analyze past states of user accounts. To address this, we'll use Debezium and ClickHouse to track and store all account-type changes with timestamps.

Setting Up Your Environment

Getting the project from Git

To get started, clone the postgres-cdc-clickhouse GitHub repository and launch the environment using Docker Compose:

git clone https://github.com/ranjbaryshahab/postgres-cdc-clickhouse
cd postgres-cdc-clickhouse
docker-compose up

Creating the PostgreSQL Table

First, define the users' table in PostgreSQL to store user data and track changes:

CREATE TABLE users (
    user_id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    account_type VARCHAR(20) NOT NULL,
    updated_at TIMESTAMP DEFAULT timezone('UTC', CURRENT_TIMESTAMP),
    created_at TIMESTAMP DEFAULT timezone('UTC', CURRENT_TIMESTAMP)
);

ALTER TABLE public.users REPLICA IDENTITY FULL;

then inserting data:

INSERT INTO users (username, account_type) VALUES
('user1', 'Bronze'),
('user2', 'Silver'),
('user3', 'Gold');

and the result is like:

Setting Up Debezium Connector

Configure the Debezium PostgreSQL connector in Kafka Connect to capture changes from the users table:

curl --location --request POST 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
    "name": "raw_data.shop-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.dbname": "postgres",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.shop",
        "database.hostname": "postgres",
        "database.password": "postgres",
        "database.port": "5432",
        "database.server.name": "shop",
        "database.user": "postgres",
        "name": "raw_data.shop-connector",
        "plugin.name": "pgoutput",
        "table.include.list": "public.users",
        "tasks.max": "1",
        "topic.creation.default.cleanup.policy": "delete",
        "topic.creation.default.partitions": "1",
        "topic.creation.default.replication.factor": "1",
        "topic.creation.default.retention.ms": "604800000",
        "topic.creation.enable": "true",
        "topic.prefix": "raw_data.shop",
        "database.history.skip.unparseable.ddl": "true",
        "key.converter": " org.apache.kafka.connect.json.JsonConverter",
        "value.converter": " org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "false",
        "value.converter.schemas.enable": "false",
        "snapshot.mode": "initial"
    }
}'

see the result in Redpanda Console:
Go to http://localhost:9080/topics to see the topics created by the Debezium connector.

Tracking Changes in Real-Time

Now, let's simulate an update in the users' table to see the change reflected in our Kafka topics and ClickHouse:

-- Update user with ID 1 to change the account type to 'Gold'
UPDATE users
SET account_type = 'Gold', updated_at = timezone('UTC', CURRENT_TIMESTAMP)
WHERE user_id = 1;

When this update is made, Debezium captures the change and produces an event in the Kafka topic. This event contains the previous and new values of the account_type field along with the operation type (u for update).

see the result in Redpanda Console:

Storing Changes in ClickHouse

Utilize ClickHouse to store all changes in a dedicated table and materialized view:

Create ClickHouse Database and Table

CREATE DATABASE shop;

CREATE TABLE shop.account_type_switch (
    user_id UInt32,
    username String,
    before_account_type String,
    after_account_type String,
    updated_at DateTime,
    created_at DateTime
) ENGINE = ReplacingMergeTree
ORDER BY (user_id, updated_at)
SETTINGS index_granularity = 8192;

Set Up Kafka Engine Table in ClickHouse

CREATE DATABASE kafka_shop;

CREATE TABLE kafka_shop.kafka__account_type_switch (
    `jsonString` String
) ENGINE = Kafka
SETTINGS kafka_broker_list = 'broker:29092',
        kafka_topic_list = 'raw_data.shop.public.users', 
        kafka_group_name = 'raw_date_clickhouse',
        kafka_format = 'JSONAsString';

Create Materialized View

CREATE MATERIALIZED VIEW kafka_shop.consumer__account_type_switch TO shop.account_type_switch (
    user_id UInt32,
    username String,
    before_account_type String,
    after_account_type String,
    updated_at DateTime,
    created_at DateTime
) AS 
SELECT
    JSONExtract(JSONExtract(jsonString, 'after', 'Nullable(String)'), 'user_id', 'Nullable(UInt32)') AS user_id,
    JSONExtract(JSONExtract(jsonString, 'after', 'Nullable(String)'), 'username', 'Nullable(String)') AS username,
    JSONExtract(JSONExtract(jsonString, 'before', 'Nullable(String)'), 'account_type', 'Nullable(String)') AS before_account_type,
    JSONExtract(JSONExtract(jsonString, 'after', 'Nullable(String)'), 'account_type', 'Nullable(String)') AS after_account_type,
    toDateTime(JSONExtract(JSONExtract(jsonString, 'after', 'Nullable(String)'), 'updated_at', 'Nullable(UInt64)') / 1000000) AS updated_at,
    toDateTime(JSONExtract(JSONExtract(jsonString, 'after', 'Nullable(String)'), 'created_at', 'Nullable(UInt64)') / 1000000) AS created_at
FROM kafka_shop.kafka__account_type_switch
WHERE (before_account_type != after_account_type) AND (JSONExtract(jsonString, 'op', 'Nullable(String)') = 'u');

Verify the data in ClickHouse

Conclusion

Implementing CDC with Debezium and ClickHouse enables your e-commerce platform to maintain a complete user account type change history. This setup empowers detailed auditing, regulatory compliance, and advanced analytics, providing valuable insights into user behavior and business trends.

By following these steps, you can enhance your data management practices and leverage real-time change tracking to drive informed decision-making in your business.

Storing and Handling Confidential Data in ClickHouse

Shahab Ranjbary — Sat, 01 Jun 2024 03:42:02 +0000

Introduction

Handling sensitive data securely is critical, especially when migrating from one database to another. In this article, we'll explore how to manage and store confidential data in ClickHouse, focusing on a scenario where we migrate data from MySQL daily. We will use encryption to ensure data security and implement views to control access.

Scenario

We have a MySQL table containing users' general information, including confidential fields like phone numbers and national codes. Our goal is to migrate this data to ClickHouse every day while ensuring the sensitive fields are encrypted.

MySQL Table Setup

First, let's set up our MySQL database and table:

CREATE DATABASE users;

CREATE TABLE users.general_info (
    id INT,
    first_name VARCHAR(255),
    last_name VARCHAR(255),
    phone_number VARCHAR(255),
    national_code VARCHAR(255)
);

INSERT INTO users.general_info VALUES
(1, 'Shahab', 'Ranjbary', '0912345678', '123456'),
(2, 'Sepehr', 'Ranjbary', '0935999999', '982345'),
(3, 'Mostafa', 'Fekri', '09192222222', '3244442'),
(4, 'Saber', 'Farshbaf', '0912333435', '454554');

ClickHouse Table Setup

Next, we create a similar table in ClickHouse with an additional column for the initialization vector (IV) used in encryption:

CREATE DATABASE users;

CREATE TABLE users.general_info (
    id INT,
    first_name VARCHAR(255),
    last_name VARCHAR(255),
    phone_number VARCHAR(255),
    national_code VARCHAR(255),
    iv VARCHAR(255)
)
ENGINE = MergeTree
PRIMARY KEY id
ORDER BY id
SETTINGS index_granularity = 8192;

Data Migration and Encryption

We'll use the encrypt function to insert data into ClickHouse. This function will encrypt the confidential fields using AES-256-GCM encryption. You can automate this process using Airflow or other scheduling tools.

INSERT INTO users.general_info
SELECT 
    id, 
    encrypt('aes-256-gcm', first_name, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS first_name,
    encrypt('aes-256-gcm', last_name, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS last_name,
    encrypt('aes-256-gcm', phone_number, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS phone_number,
    encrypt('aes-256-gcm', national_code, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS national_code,
    'iv1' AS iv
FROM mysql('mysql:3306','users', 'general_info', 'root','123456');

Real-Time Data Handling

If you use Kafka, NATS, or another streaming service, you can use a materialized view in ClickHouse to encrypt data upon ingestion.

CREATE MATERIALIZED VIEW users.general_info_mv TO users.general_info AS
SELECT 
    id, 
    encrypt('aes-256-gcm', first_name, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS first_name,
    encrypt('aes-256-gcm', last_name, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS last_name,
    encrypt('aes-256-gcm', phone_number, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS phone_number,
    encrypt('aes-256-gcm', national_code, 'keykeykeykeykeykeykeykeykeykey01', 'iv1') AS national_code,
    'iv1' AS iv
FROM kafka('kafka_broker:9092', 'topic', 'group', 'format');

Decryption and Secure Access

For users who need access to decrypted data, we create a view that decrypts the confidential fields on the fly using the tryDecrypt function.

CREATE VIEW users.general_info_decryption AS
SELECT 
    id,
    tryDecrypt('aes-256-gcm', first_name, 'keykeykeykeykeykeykeykeykeykey01', iv) AS first_name,
    tryDecrypt('aes-256-gcm', last_name, 'keykeykeykeykeykeykeykeykeykey01', iv) AS last_name,
    tryDecrypt('aes-256-gcm', phone_number, 'keykeykeykeykeykeykeykeykeykey01', iv) AS phone_number,
    tryDecrypt('aes-256-gcm', national_code, 'keykeykeykeykeykeykeykeykeykey01', iv) AS national_code
FROM users.general_info;

Managing Permissions

We can control data access by granting permissions selectively to users:

-- Grant access to the main table for all users
GRANT SELECT ON users.general_info TO public;

-- Grant access to the decrypted view only to specific users
GRANT SELECT ON users.general_info_decryption TO admin;

Conclusion

We can securely handle confidential data by leveraging ClickHouse's encryption functions and carefully managing user access. This approach ensures that sensitive information remains protected during storage and transit, while still being accessible to authorized users.

Additional Tips

Key Management: Use a secure key management system (KMS) to store and rotate encryption keys. Avoid hardcoding keys in your scripts or configurations.
Auditing: Implement logging and monitoring to audit access to encrypted and decrypted data.
Performance: Be mindful of the performance implications of encryption and decryption. Test your setup to ensure it meets your performance requirements.

By following these best practices, you can enhance the security and compliance of your data handling processes in ClickHouse.

For more information on ClickHouse's encryption functions and best practices, refer to the ClickHouse documentation.

Universal Data Migration: Using Slingdata to Transfer Data Between Databases

Shahab Ranjbary — Fri, 24 May 2024 12:35:51 +0000

Overview

Data management can be complex, especially when dealing with high-volume data pipelines. Whether you're moving data between databases, from file systems to databases, or vice versa, the Extract & Load (EL) approach can streamline these processes. Enter Slingdata, a powerful, free CLI tool built with Go that offers a straightforward solution for creating and maintaining robust data pipelines.

What is Slingdata?

Slingdata is a passion project turned into a practical tool that simplifies data movement across various systems. Its key focus is on the Extract & Load (EL) process, enabling efficient data transfer between:

Database to Database
File System to Database
Database to File System

Key Features of Slingdata

Single Binary deployment (built with Go). See the installation page.
Use Custom SQL as a stream.
Manage / View / Test / Discover your connections with the sling conns sub-command.
Use Environment Variables for connections.
Provide YAML or JSON configurations (perfect for git version control).
Powerful Replication logic, to replicate many tables with a wildcard (my_schema.*).
Reads your existing DBT connections.
Use your environment variable in your YAML / JSON config (SELECT * from my_table where date = '{date}').
Convenient Transformations, such as the flatten option, which auto-creates columns from your nested fields.
Run Pre & Post SQL commands.
And many more!

Sample EL Process: ClickHouse to PostgreSQL

To showcase Slingdata’s capabilities, let's walk through a sample Extract & Load process moving data from ClickHouse to PostgreSQL.

Prerequisites

Ensure you have:

ClickHouse installed and running.
PostgreSQL installed and running.
Slingdata installed.

Step 1: Configure the Source Database

Let's assume our source database is ClickHouse. We can export a new environment variable or add the connection to the env.yaml of Slingdata files:

export MARKETING_URL="postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

To test this connection:

sling conns test MARKETING_URL

You should see the result:

INF success!

We will do the same for ClickHouse:

export DATAWAREHOUSE_URL="http://sling_user:sling_pass@clickhouse:8123/marketing"

To test this connection:

sling conns test DATAWAREHOUSE_URL

You should see the result:

INF success!

Now, let's create a database and a table in ClickHouse:

create database marketing;

CREATE TABLE `marketing`.users (
    id UInt64,
    name String,
    family String,
    age UInt8
) ENGINE = MergeTree()
ORDER BY id;

And in PostgreSQL, create a table with the same schema:

create table users(
   id serial not null primary key,
   name text not null,
   family text,
   age text not null
);

Insert data into ClickHouse:

insert into marketing.users values(1,'Shahab', 'Ranjbary', 28);
insert into marketing.users values(2,'Sepher', 'Ranjbary', 18);
insert into marketing.users values(3,'Mostafa', 'Fekri', 38);

Now, to move data to PostgreSQL, you can either create a replication file or run the following command:

sling run --src-conn DATAWAREHOUSE_URL --src-stream marketing.users --tgt-conn MARKETING_URL --tgt-object public.users

Conclusion

Slingdata offers a powerful and flexible way to manage data pipelines with ease. Its extensive features, combined with simple configurations and robust performance, make it an excellent tool for anyone dealing with high-volume data transfers.

Simplified API Creation and Management: ClickHouse to APISIX Integration Without Code

Shahab Ranjbary — Wed, 22 May 2024 19:23:51 +0000

Introduction

In the world of data management and web services, creating and managing APIs can often be a complex and time-consuming task. However, with the right tools, this process can be significantly simplified. In this article, we will explore how to create APIs for fetching data from ClickHouse tables without writing any code and manage these APIs using APISIX. ClickHouse, a fast and open-source columnar database management system, provides an HTTP interface by default, enabling easy access to data. By integrating this with APISIX, an open-source API gateway, we can not only manage and log our APIs but also leverage a host of features provided by APISIX to enhance our API management capabilities.

Why ClickHouse and APISIX?

ClickHouse is renowned for its performance and efficiency in handling large volumes of data. Its built-in HTTP interface allows for straightforward data retrieval from tables, making it an excellent choice for building APIs.

APISIX, on the other hand, is a powerful API gateway that offers extensive features such as traffic management, logging, monitoring, and security, among others. Integrating ClickHouse with APISIX allows us to manage our APIs effectively without delving into the complexities of writing custom code.

Creating APIs in ClickHouse

ClickHouse’s HTTP interface enables us to interact with the database using simple HTTP requests. This interface supports various operations such as querying data, inserting records, and managing tables. To fetch data from a ClickHouse table, you can use a basic HTTP GET request:

GET http://<clickhouse-server>:8123/?query=SELECT+*+FROM+<table_name>

This simplicity makes ClickHouse an excellent candidate for API creation, as it eliminates the need for complex server-side logic. However, managing these APIs, especially in a production environment, requires additional capabilities such as authentication, rate limiting, and detailed logging.

Enhancing API Management with APISIX

APISIX comes into play by providing a robust platform for managing our ClickHouse APIs. By routing ClickHouse's HTTP requests through APISIX, we gain access to a plethora of features that enhance API management:

Traffic Management: Control and optimize API traffic using load balancing and rate limiting.
Security: Implement authentication and authorization mechanisms to secure your APIs.
Logging and Monitoring: Gain insights into API usage patterns and performance through comprehensive logging and monitoring tools.
Plugin System: Extend APISIX’s functionality using its rich plugin system, enabling custom behavior and integrations.

Setting Up the Integration

To run, follow these steps:

Clone the Repository

First, clone the repository to your local machine. This repository contains the necessary configurations for setting up ClickHouse and APISIX.

git clone https://github.com/ranjbaryshahab/clickhouse-apisix.git
cd clickhouse-apisix

Run Docker Compose

Start the services using Docker Compose. This command will set up ClickHouse and APISIX using predefined configurations.

docker-compose up -d

Connect to ClickHouse and Run the Following Commands

Create table

Create a table named users in the default database of ClickHouse. This table will store user data.

CREATE TABLE `default`.users (
    id UInt64,
    name String,
    family String,
    age UInt8
) ENGINE = MergeTree()
ORDER BY id;

Insert data into the table

Insert sample data into the users table. This data will be used for testing our API.

insert into users values(1, 'Shahab', 'Ranjbary',28);
insert into users values(2, 'Sepehr', 'Ranjbary',18);
insert into users values(3, 'John', 'Doe',28);

Create a Log Table for the API Calls

Create a log table in the apisix_log database. This table will store logs of API calls for monitoring and analysis.

CREATE database apisix_log;

CREATE TABLE apisix_log.users
(
    `@timestamp` String,
    `host` String,
    `client_ip` String,
    `consumer_name` String,
    `route_id` String,
    `route_name` String,
    `request` String,
    `upstream_status` String,
    `status` UInt16,
    `upstream_response_time` Float64,
    `connection_time` Float64
)
ENGINE = MergeTree
PRIMARY KEY `@timestamp`
ORDER BY `@timestamp`
SETTINGS index_granularity = 8192

Create Users and Grant Access

Create users for APISIX and grant the necessary permissions to access the users table and insert logs into the apisix_log.users table.

CREATE USER apisix IDENTIFIED WITH plaintext_password BY 'apisix';
GRANT SHOW COLUMNS, SELECT on default.users TO apisix;

CREATE USER apisix_logger IDENTIFIED WITH plaintext_password BY 'apisix_logger';
GRANT SHOW COLUMNS, SELECT, INSERT on apisix_log.users TO apisix_logger;

Create a Consumer in APISIX

Create a consumer in APISIX with basic authentication. This consumer will be used to authenticate API requests.

curl http://localhost:9180/apisix/admin/consumers \
-H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '
{
  "username": "ds_team",
  "plugins": {
    "basic-auth": {
      "_meta": {
        "disable": true
      },
      "password": "12345",
      "username": "ds_team"
    }
  }
}'

Create a Route in APISIX

Create a route in APISIX to handle requests to the ClickHouse API. This route includes configurations for authentication, logging, and proxying requests to ClickHouse.

curl http://localhost:9180/apisix/admin/routes/1 \
-H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '
{
  "uri": "/clickhouse-api*",
  "name": "clickhouse",
  "methods": [
    "GET"
  ],
  "plugins": {
    "basic-auth": {
      "disable": false
    },
    "clickhouse-logger": {
      "database": "apisix_log",
      "disable": false,
      "endpoint_addr": "http://clickhouse:8123",
      "log_format": {
        "@timestamp": "$time_iso8601",
        "client_ip": "$remote_addr",
        "connection_time": "$connection_time",
        "consumer_name": "$consumer_name",
        "host": "$host",
        "request": "$request",
        "route_name": "$route_name",
        "status": "$status",
        "upstream_response_time": "$upstream_response_time",
        "upstream_status": "$upstream_status"
      },
      "logtable": "users",
      "password": "apisix_logger",
      "user": "apisix_logger"
    },
    "proxy-rewrite": {
      "headers": {
        "Authorization": "",
        "X-ClickHouse-Format": "JSON",
        "X-ClickHouse-Key": "apisix",
        "X-ClickHouse-User": "apisix"
      },
      "uri": "/?query=select%20name,family,age%20from%20default.users%20where%20age%20=%20%7Bage:UInt8%7D"
    }
  },
  "upstream": {
    "nodes": [
      {
        "host": "clickhouse",
        "port": 8123,
        "weight": 1
      }
    ],
    "timeout": {
      "connect": 6,
      "send": 6,
      "read": 6
    },
    "type": "roundrobin",
    "scheme": "http",
    "pass_host": "pass",
    "keepalive_pool": {
      "idle_timeout": 60,
      "requests": 1000,
      "size": 320
    }
  },
  "status": 1
}'

Now We Can Test It

Test the API by making a GET request to the ClickHouse API endpoint. This request fetches users with the age of 18.

curl --location --request GET 'http://localhost:9080/clickhouse-api/users?param_age=18' \
--header 'Authorization: Basic ZHNfdGVhbToxMjM0NQ=='

This API returns the users with the specified age. The result is:

{
    "meta": [
        {
            "name": "name",
            "type": "String"
        },
        {
            "name": "family",
            "type": "String"
        },
        {
            "name": "age",
            "type": "UInt8"
        }
    ],
    "data": [
        {
            "name": "Sepehr",
            "family": "Ranjbary",
            "age": 18
        }
    ],
    "rows": 1,
    "statistics": {
        "elapsed": 0.001827077,
        "rows_read": 3,
        "bytes_read": 35
    }
}

You can also check the log table for verification:

SELECT * FROM apisix_log.users;

Conclusion

Integrating ClickHouse with APISIX allows you to create and manage APIs efficiently without writing any code. ClickHouse’s HTTP interface simplifies data retrieval, while APISIX enhances API management with features like traffic control, security, and logging. This setup is highly beneficial for production environments, where managing APIs with minimal code and maximum efficiency is crucial. By following the steps outlined in this guide, you can quickly set up and test APIs for your ClickHouse tables, ensuring robust and scalable API management.

Improving Data Quality in ClickHouse Databases with Soda

Shahab Ranjbary — Fri, 19 Jan 2024 23:22:59 +0000

Introduction

Data quality is the bedrock of any data-driven project, ensuring the integrity and reliability of information. Soda, a versatile tool catering to Data Engineers, Data Scientists, and Data Analysts, empowers users to conduct data quality tests precisely when and where they need to. This guide merges theoretical insights with practical examples from the soda-clickhouse project, showcasing how Soda can significantly enhance data quality in ClickHouse databases.

Soda Use Case Guides

1. Testing Data in a Pipeline

Integrate Soda seamlessly into your data pipeline to perform continuous data quality checks. Define checks using SodaCL, include scans at key pipeline stages, and monitor results for early issue detection, ensuring data quality at every step.

2. Testing Data Before Migration

Prioritize data integrity before migration by configuring Soda for both source and target databases. Leverage Soda's reconciliation checks to verify data consistency and resolve any issues identified in pre-migration scans, ensuring a smooth transition.

3. Testing Data in CI/CD

Incorporate Soda into your CI/CD pipeline for automated data quality checks. Configure alerts for failures, seamlessly integrate with GitHub Actions, and maintain reliable data throughout the development lifecycle, fostering a culture of quality in your data-driven processes.

4. Self-Serve Soda

Empower your teams with self-serve capabilities using Soda. Facilitate easy access, encourage collaborative definition of quality checks, enable browser interface usage, and integrate with data catalogs for a holistic overview of dataset health, allowing teams to take ownership of their data quality.

Soda Cloud: Advanced Analytics and Collaboration

When it comes to reviewing scan results and investigating issues, Soda Cloud takes your data quality management to the next level. Here's what you can do:

Review Scan Results:
- Access visualized scan results not only in the command-line interface but also through the intuitive dashboard on Soda Cloud.
Set Alert Notifications:
- Configure custom alert notifications to stay informed about any deviations or anomalies detected during scans. Proactively address potential data quality issues.
Track Trends Over Time:
- Gain insights into the trends of your data quality metrics over time. Track improvements or deviations and make informed decisions about your data pipeline.
Integration with External Tools:
- Seamlessly integrate Soda Cloud with your existing messaging, ticketing, and data cataloging tools. Collaborate more effectively by connecting with platforms like Slack, Jira, and Atlan.

Soda Cloud not only enhances the visibility of your data quality but also facilitates collaboration and advanced analytics, making it a powerful companion for managing and improving your data quality standards.

Project Samples: `soda-clickhouse`

Explore the soda-clickhouse project on GitHub, featuring two insightful samples:

Sample_1: Data Quality in dim_customer Table
- Dive into practical examples of performing quality checks on the dim_customer table within ClickHouse using Soda.
Sample_2: Data Migration Validation
- Ensure data integrity during migration from a MySQL to ClickHouse data source using reconciliation checks in Soda.

Conclusion

Implementing Soda for data quality assurance provides a comprehensive and effective approach. Whether you are testing data in a pipeline, before migration, in CI/CD, or enabling self-serve capabilities, Soda enhances the reliability and trustworthiness of your datasets. Explore the soda-clickhouse project, experiment with the provided samples, and leverage Soda's capabilities to elevate your data quality standards and foster a data-driven culture. Elevate your data quality standards with Soda - making data reliability an integral part of your data journey.

Using Benthos: A Practical Guide for Kafka and PostgreSQL Integration

Shahab Ranjbary — Fri, 08 Dec 2023 23:12:09 +0000

Introduction

Benthos is a versatile and powerful stream processor that simplifies the task of handling real-time data streams. In this article, we'll delve into Benthos, its key features, and how it can be seamlessly integrated with Kafka and PostgreSQL to build robust data pipelines. To illustrate these concepts, we'll use a practical example: BenthosPipelineDB.

BenthosPipelineDB is a sample project showcasing the seamless integration of Benthos with Kafka and PostgreSQL. This project is designed to produce random data, publish it to a Kafka topic, and consume and insert it into a PostgreSQL database using Benthos. The project structure and configurations provide a clear understanding of how these technologies can be orchestrated to create a reliable and scalable data processing pipeline.

Benthos Overview

Benthos is a stream processor designed for real-time data handling. Its features include:

Stream Processing: Benthos allows you to process data streams with ease, making it a valuable tool for scenarios that require real-time data transformations.
Connectivity: With native support for various data sources and sinks, including Kafka and PostgreSQL, Benthos simplifies the integration process.
Extensibility: Benthos supports a wide range of processors and plugins, providing flexibility in designing your data processing pipeline.

Kafka Overview

Kafka is a distributed streaming platform that enables the building of real-time data pipelines. Key Kafka concepts include:

Topics: Kafka organizes data streams into topics, allowing for efficient data segregation and distribution.
Producers and Consumers: Producers publish data to Kafka topics, while consumers subscribe to these topics, creating a scalable and fault-tolerant system.

PostgreSQL Overview

PostgreSQL is a powerful, open-source relational database system. It offers:

Relational Model: PostgreSQL follows a robust relational database model, supporting the creation of structured and organized datasets.
ACID Compliance: ACID (Atomicity, Consistency, Isolation, Durability) compliance ensures data integrity, making PostgreSQL suitable for critical applications.

Project Structure

Here's an overview of the project structure:

Benthos Pipeline: The core Benthos configuration is stored in pipeline/benthos.yml.
PostgreSQL Setup: SQL scripts for creating tables are located in postgres/sql/create_table.sql, and data is stored in postgres/data.
Data Producer: The data generator, responsible for producing random data, is in the data-producer directory.
Docker Setup: The docker-compose.yaml file orchestrates the Docker containers for PostgreSQL, Kafka, Benthos, and the data producer.
Configuration: Environment variables are stored in config.env, facilitating easy configuration management.

How It Works

Data Generation: The data producer (message-producer.py) generates random messages with the format {"name": "hRnWJsIf", "age": 82} and publishes them to the Kafka topic (users).
Benthos Processing: Benthos reads messages from the Kafka topic, processes them according to the defined pipeline in pipeline/benthos.yml, and sends them to the specified output.
PostgreSQL Storage: Processed data is inserted into the PostgreSQL table (users) as defined in postgres/sql/create_table.sql.

Running the Project

1- Navigate to the project directory:

cd /path/to/BenthosPipelineDB

2- Start the project using Docker Compose:

docker-compose up -d

3- Monitor logs to ensure everything is running smoothly:

docker-compose logs -f

4- Kafka Console Consumer
If you want to observe the data flowing through the users topic in real time, you can use the Kafka console consumer. Open your terminal and run the following command:

docker-compose exec kafka kafka-console-consumer.sh --topic users --from-beginning --bootstrap-server kafka:9092

5- Connecting to PostgreSQL
To inspect the data in the PostgreSQL database, you can use a PostgreSQL client. Assuming you have PostgreSQL installed locally, you can connect using the following command:

psql -h localhost -p 5432 -U postgres -d postgres

6- Now, let's run a simple query to fetch the first 10 records from the users' table:

SELECT * FROM users LIMIT 10;

Now, you have a robust data processing pipeline using Benthos, Kafka, and PostgreSQL!

Conclusion

BenthosPipelineDB demonstrates the power and flexibility of Benthos in combination with Kafka and PostgreSQL. With a clear project structure and straightforward configuration, it provides a foundation for building scalable and reliable data processing systems.

Explore the BenthosPipelineDB repository for hands-on experience and customization.

Happy streaming!

Connecting Multiple Kafka Clusters in ClickHouse Using Named Collections

Shahab Ranjbary — Mon, 25 Sep 2023 00:47:53 +0000

Introduction:

ClickHouse is a powerful columnar database renowned for its speed and efficiency. A pivotal strength lies in its seamless integration with external data sources like Kafka. With the rising need for multi-cluster setups in modern data architectures, ClickHouse's Named Collections offers an invaluable asset. In this guide, we'll delve into how you can leverage this feature to seamlessly set up connections to two distinct Kafka clusters.

Why Use Named Collections?

Understanding the true value of Named Collections is crucial before we dive deep into the configurations. They allow us to:

Reduce Repetition: Eliminate the need to redundantly specify configurations.
Centralized Management: Maintain all configurations in a single, easily manageable location.
Improved Security: Safeguard sensitive credentials, keeping them out of the reach of non-administrative users.

Configuring Named Collections for Kafka:

With the prominence of Named Collections established, let's gear up to connect to two distinct Kafka clusters – primary and secondary.

XML Configuration:

<clickhouse>
    <named_collections>
        <!-- Primary Kafka Cluster Configuration -->
        <primary_kafka_cluster>
            <broker_list>primary-kafka-cluster:9094</broker_list>
            <kafka_settings>
                <client_id>primary_kafka_client</client_id>
                <security_protocol>SASL_PLAINTEXT</security_protocol>
                <sasl_mechanisms>SCRAM-SHA-512</sasl_mechanisms>
                <sasl_username>clickhouse_primary</sasl_username>
                <sasl_password>primary_secret_password</sasl_password>
            </kafka_settings>
        </primary_kafka_cluster>
        <!-- Secondary Kafka Cluster Configuration -->
        <secondary_kafka_cluster>
            <broker_list>backup-kafka-cluster:9095</broker_list>
            <kafka_settings>
                <client_id>secondary_kafka_client</client_id>
                <security_protocol>SASL_PLAINTEXT</security_protocol>
                <sasl_mechanism>SCRAM-SHA-512</sasl_mechanism>
                <sasl_username>clickhouse_secondary</sasl_username>
                <sasl_password>secondary_secret_password</sasl_password>
            </kafka_settings>
        </secondary_kafka_cluster>
    </named_collections>
</clickhouse>

For a more detailed configuration setup, refer to Pull Request #31691 starting from ClickHouse v21.12, which provides a more streamlined approach to using named_collections.

Setting Up Permanent Storage: MergeTree Table

After configuring our Kafka connections, the focus shifts to the ClickHouse realm. We'll architect tables that act as our permanent data reservoirs.

1. Kafka Engine Table:

To tap directly into our Kafka topics, we'll shape tables in ClickHouse using the Kafka engine. Here's how you can define these tables:

For the primary Kafka cluster:

CREATE TABLE kafka_cluster_a
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
)
ENGINE = Kafka(primary_kafka_cluster)
SETTINGS kafka_topic_list = 'your_topic_name_for_primary',
         kafka_group_name = 'your_consumer_group_for_primary',
         kafka_format = 'JSONEachRow';

For the secondary Kafka cluster:

CREATE TABLE kafka_cluster_b
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
)
ENGINE = Kafka(secondary_kafka_cluster)
SETTINGS kafka_topic_list = 'your_topic_name_for_secondary',
         kafka_group_name = 'your_consumer_group_for_secondary',
         kafka_format = 'JSONEachRow';

2. MergeTree Table:

We'll use the MergeTree table to persistently store the data streamed from Kafka:

For kafka.cluster_a:

CREATE TABLE cluster_a_storage
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
) ENGINE = MergeTree()
ORDER BY id;

For kafka.cluster_b:

CREATE TABLE cluster_b_storage
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
) ENGINE = MergeTree()
ORDER BY id;

3. Materialized View:

The Materialized View serves as the Kafka table's consumer, directing data flow:

For kafka.cluster_a:

CREATE MATERIALIZED VIEW cluster_a_mv TO cluster_a_storage AS
SELECT 
    id,
    first_name,
    last_name
FROM kafka.cluster_a;

For kafka.cluster_b:

CREATE MATERIALIZED VIEW cluster_b_mv TO cluster_b_storage AS
SELECT 
    id,
    first_name,
    last_name
FROM kafka.cluster_b;

Practical Applications:

With the above groundwork, ClickHouse is primed to consistently ingest and archive data from both Kafka clusters. This means any data dispatched to the delineated Kafka topics will be assimilated in real time. This is particularly advantageous for businesses seeking to conduct instantaneous analytics or data-driven decision-making.

Conclusion:

Harnessing ClickHouse's Named Collections, establishing connections to multiple Kafka clusters transitions from being merely possible to efficient and organized. This structure guarantees instant data availability for querying, simplifying real-time analytics.

DEV Community: Shahab Ranjbary

Immutable Data ClickHouse, Mutable Data StarRocks: Why I Moved My CDC Tables

The problem: mutable data

Why ClickHouse was painful for this job

1. "Current state" is eventual, not immediate

2. Deletes are a flag, not a delete

3. Joins across many mutable tables

4. Three objects per table

5. Small Avro surprises

Why StarRocks fits mutable data

Primary Key tables: upsert and delete at write time

Conditional updates for back-fills

It speaks MySQL

Joins are a first-class citizen

Partial updates and schema changes

The cons (the honest part)

1. Silent NULLs with strict_mode = false

2. Your primary key must never change at the source

3. merge_condition can't protect against a delete

4. The primary key index costs memory

5. Throughput scales with backends

6. Smaller ecosystem

7. Now there are two databases to run

Side by side

Where we ended up

Checklist if you're doing the same

From SSH Chaos to Self-Service GPU Workspaces with Coder and HAMi

The problem: one GPU box, many data scientists, endless conflicts

The false starts (that taught me what I actually needed)

Enter Coder

What I added to Coder OSS

Building a fork with AI assistance

Then the GPUs fought back

HAMi to the rescue

Where it landed

Open source

What I'd tell you if you're doing this

🧠 ClickHouse LEFT JOINs: Why join_use_nulls Matters

🧠 Understanding join_use_nulls in ClickHouse

🧩 The Setup

🧮 The Default JOIN

🧩 Enter join_use_nulls

⚙️ Why This Matters

🚀 Performance Note

🔍 Check Your Settings

📚 Reference

🗨️ Discussion

🛠️ Copying and Moving Data Between Tables in ClickHouse (with Different Engines)

❌ Why MOVE PARTITION TO TABLE Didn’t Work for Us

✅ Alternative: ATTACH PARTITION FROM

✅ Requirements:

🧹 Final Step: Drop the Old Partition

🧾 What About ATTACH TABLE AS REPLICATED?

Why?

🧪 Summary: Best Way to Copy Between Tables with Different Engines

📚 Resources

Choosing the Right Real-Time Stream Processing Framework

Introduction

ksqlDB

Use Cases

Pros

Cons

Apache Spark (Structured Streaming)

Use Cases

Pros

Cons

Apache Flink

Use Cases

Pros

Cons

Timeplus Proton

Use Cases

Pros

Cons

RisingWave

Use Cases

Pros

Cons

Apache Storm

Use Cases

1. Silent NULLs with `strict_mode = false`

3. `merge_condition` can't protect against a delete

🧠 Understanding `join_use_nulls` in ClickHouse

❌ Why `MOVE PARTITION TO TABLE` Didn’t Work for Us

✅ Alternative: `ATTACH PARTITION FROM`

🧾 What About `ATTACH TABLE AS REPLICATED`?

Project Samples: `soda-clickhouse`