DEV Community: Shahab Ranjbary

🧠 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.

Safeguarding Secrets in Spring Boot with Vault

Shahab Ranjbary — Thu, 14 Sep 2023 15:44:16 +0000

When developing Spring Boot applications, we often encounter the need to manage sensitive information like database passwords, application keys, and other secrets. Typically, these secrets might reside in application.properties or application.yml. Pushing such sensitive information to repositories is risky. Thankfully, there's HashiCorp's Vault.

Understanding Vault

Vault is a tool for securely accessing secrets. A secret is anything that you want to tightly control access to, such as API keys, passwords, or certificates. Vault provides a unified interface to any secret while providing tight access control and recording a detailed audit log.

Setting Up Vault with Docker Compose

For rapid deployment, especially during development, Docker Compose offers a convenient method to get Vault up and running.

1.Create a file named docker-compose.yml and add the following:

version: '3'

services:
  vault:
    image: vault:1.13.3
    container_name: vault_dev
    ports:
      - "8200:8200"
    environment:
      VAULT_DEV_ROOT_TOKEN_ID: myroot
      VAULT_DEV_LISTEN_ADDRESS: "0.0.0.0:8200"
    cap_add:
      - IPC_LOCK
    command: server -dev

2.Run docker-compose up to start the Vault server.

3.Access the Vault UI at http://localhost:8200/ui and log in with the token myroot.

After logging in, you can view this page

Enable KV engine

Add the secrets in the specific path you want eg Postgresql settings

Integrating Vault with Spring Boot

Once Vault is up and running, the next step is integrating it with your Spring Boot application.

1.Setting Environments

export VAULT_URI=http://localhost:8200
export VAULT_TOKEN=myroot

2.Spring Boot Initial Setup:
Set up a basic Spring Boot application. If you're new to this, Spring Initializr provides a quick way to bootstrap a new Spring Boot project.

3.Maven Dependency:
To integrate Spring Vault, add the following dependency to your Maven pom.xml:

<dependency>
    <groupId>org.springframework.vault</groupId>
    <artifactId>spring-vault-core</artifactId>
    <version>3.0.4</version>
</dependency>

4.Integration Code:
First, define your Vault configuration:

public class VaultConfiguration {
    private static final String token;
    private static final String uri;

    static {
        token = System.getenv("VAULT_TOKEN");
        uri = System.getenv("VAULT_URI");
    }

    public static String getToken() {
        return token;
    }

    public static String getUri() {
        return uri;
    }
}

Then, use a utility to read secrets from Vault:

public class VaultConfigReaderUtil {
    public static Properties read(String path) {
        String MAIN_VAULT_PATH = "kv/data/myapplication";
        VaultEndpoint endpoint;
        Properties properties = new Properties();

        try {
            endpoint = VaultEndpoint.from(new URI(VaultConfiguration.getUri()));
            VaultTemplate vaultTemplate = new VaultTemplate(endpoint, new TokenAuthentication(VaultConfiguration.getToken()));
            VaultResponse response = vaultTemplate.read("%s/%s".formatted(MAIN_VAULT_PATH, path));
            Map<String, Object> mapData = (Map<String, Object>) response.getRequiredData().get("data");
            properties.putAll(mapData);
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }

        return properties;
    }
}

Lastly, in your main application, use the utility to fetch secrets and feed them to Spring Boot:

@SpringBootApplication
public class MyApplication {
    public static void main(String[] args) {
        Properties props = new Properties();
        Properties postgresProp = VaultConfigReaderUtil.read("postgres");
        props.putAll(postgresProp);

        SpringApplicationBuilder springApplication = new SpringApplicationBuilder(MyApplication.class);
        springApplication.properties(props).run(args);
    }
}

Conclusion

In the ever-evolving landscape of software development, security remains paramount. Secrets management is a fundamental aspect of ensuring that our applications are robust against breaches and unforeseen vulnerabilities. As we've explored, Vault by HashiCorp offers a comprehensive solution to this challenge, seamlessly integrating with frameworks like Spring Boot.

Utilizing Docker Compose, we've streamlined the deployment of Vault, making it more accessible, especially during the development phase. However, it's essential to understand that while development configurations simplify processes, they aren't suitable for production environments. Always ensure a secure, production-ready configuration for Vault when deploying in live scenarios.

Spring Boot's flexibility and vast ecosystem, paired with Vault's secure secrets management capabilities, provide developers with powerful tools to build secure, scalable, and efficient applications. As we continue to innovate and build, let's prioritize the safety and integrity of our applications, ensuring that our secrets remain just that – secret.

ClickHouse Custom Function: Convert Georgian Dates to Jalali Dates

Shahab Ranjbary — Thu, 07 Sep 2023 10:50:13 +0000

Dates are a fundamental part of data analysis, and often, there's a need to convert between different date systems. While the Gregorian (or Georgian) calendar is internationally the most widely used civil calendar, the Jalali or Persian calendar is used predominantly in certain regions like Iran. and has its own significance. Especially when dealing with data from regions using the Jalali calendar, converting between these date systems becomes essential.

In this guide, we'll walk through setting up a simple integration in ClickHouse that allows you to convert Georgian dates to Jalali dates. ClickHouse, a fast and efficient column-oriented database management system, allows the integration of custom functions. We'll leverage this capability to use a Python script for our date conversion.

Follow the steps below to set up this conversion functionality:

1. Install the Necessary Python Library:

sudo pip install jdatetime

2. Scripting the Conversion:
Initiate by creating a Python script named georgian_to_jalali.py.

sudo nano /var/lib/clickhouse/user_scripts/georgian_to_jalali.py

Populate the script with:

#!/usr/bin/python3
import jdatetime
import sys

def georgian_to_jalali(date_str):
    georgian_date = jdatetime.datetime.strptime(date_str, '%Y-%m-%d')
    jalali_date = jdatetime.datetime.fromgregorian(datetime=georgian_date).strftime('%Y-%m-%d')
    return jalali_date

def main():
    for line in sys.stdin:
        date_str = str(line.rstrip())
        print(georgian_to_jalali(date_str), end='')
        sys.stdout.flush()

if __name__ == "__main__":
    main()

Ensure the script is executable:

sudo chmod 777 /var/lib/clickhouse/user_scripts/georgian_to_jalali.py

3. Integrate with ClickHouse:
Define the ClickHouse function by creating an XML file:

sudo nano /etc/clickhouse-server/georgian_to_jalali_function.xml

Input the following XML:

<functions>
    <function>
        <type>executable</type>
        <name>georgian_to_jalali</name>
        <return_type>String</return_type>
        <argument>
            <type>String</type>
            <name>value</name>
        </argument>
        <format>TabSeparated</format>
        <command>georgian_to_jalali.py</command>
    </function>
</functions>

Usage:
With everything set, utilize the function in ClickHouse queries. Sample queries include:

SELECT today() as georgian_date, georgian_to_jalali(today()) as jalali_date;

After executing, the results align with the referenced image provided.

DEV Community: Shahab Ranjbary

🧠 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

Pros

Cons

Conclusion

Additional Insights from Streaming Databases

References

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

Introduction

Scenario Overview

Setting Up Your Environment

Getting the project from Git

Creating the PostgreSQL Table

Setting Up Debezium Connector

Tracking Changes in Real-Time

Storing Changes in ClickHouse

Create ClickHouse Database and Table

Set Up Kafka Engine Table in ClickHouse

Create Materialized View

Verify the data in ClickHouse

Conclusion

Storing and Handling Confidential Data in ClickHouse

Introduction

Scenario

MySQL Table Setup

ClickHouse Table Setup

Data Migration and Encryption

Real-Time Data Handling

Decryption and Secure Access

Managing Permissions

Conclusion

Additional Tips

Universal Data Migration: Using Slingdata to Transfer Data Between Databases

Overview

What is Slingdata?

Key Features of Slingdata

Sample EL Process: ClickHouse to PostgreSQL

Prerequisites

🧠 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`