طراحی پایپلاین بلادرنگ CDC با Debezium، کافکا (Kafka) و کلیکهاوس (ClickHouse)

چرا کوئریهای دورهای (Batch Polling) در مقیاس بالا شکست میخورند؟

در معماریهای سنتی ETL، همگامسازی دیتابیس عملیاتی (OLTP) مانند PostgreSQL با انبار داده تحلیلی (OLAP) از طریق اجرای کوئریهای زمانبندیشده بر پایه ستون updated_at انجام میشود. این رویکرد در مقیاس بالا سه بحران اساسی ایجاد میکند:

• تحمیل بار شدید I/O و CPU روی دیتابیس اصلی: اجرای مداوم کوئریهای SELECT سنگین روی جداول چندصد میلیونی باعث اشغال Buffer Pool و افت سرعت تراکنشهای کاربران میشود.
• عدم توانایی در تشخیص Hard Delete: اگر رکوردی با دستور DELETE از دیتابیس حذف شود، هیچ ردی از آن در کوئریهای مبتنی بر updated_at باقی نمیماند و انبار داده دچار ناسازگاری دائمی (Data Drift) میشود.
• از دست رفتن وضعیتهای میانی (Intermediate States): اگر وضعیت یک سفارش در فاصله بین دو اجرای Cron Job سه بار تغییر کند، تنها آخرین وضعیت ثبت شده و تاریخچه تغییرات از دست میرود.

راهکار استاندارد مهندسی داده برای حل این معضل، استفاده از Log-Based CDC (Change Data Capture) و خواندن مستقیم لاگهای تراکنش دیتابیس است.

کالبدشکافی PostgreSQL WAL و مکانیسم Debezium

در PostgreSQL، هر تغییر داده (INSERT، UPDATE، DELETE) پیش از اعمال روی فایلهای جدول، به صورت ترتیبی در فایلهای WAL (Write-Ahead Log) نوشته میشود. Debezium به عنوان یک کانکتور روی Kafka Connect، با استفاده از پلاگین استاندارد pgoutput و ایجاد یک Replication Slot، استریم باینری WAL را در لحظه رمزگشایی کرده و به رویدادهای ساختاریافته تبدیل میکند.

مدیریت خطر پر شدن دیسک در Replication Slots

اگر کانکتور Debezium به هر دلیلی متوقف شود، PostgreSQL فایلهای WAL را پاک نمیکند تا زمانی که کانکتور دوباره بالا بیاید و آنها را بخواند. در دیتابیسهای پرترافیک، این موضوع میتواند در عرض چند ساعت دیسک سرور اصلی را ۱۰۰ درصد پر کرده و کل سیستم را متوقف کند. برای مهار این ریسک، تنظیم دو پارامتر زیر حیاتی است:

• تنظیم max_slot_wal_keep_size در postgresql.conf برای تعیین سقف نگهداری WAL
• فعالسازی heartbeat.interval.ms و heartbeat.action.query در تنظیمات Debezium تا در صورت کمترافیک بودن جدول هدف، اشارهگر LSN (Log Sequence Number) به صورت دورهای به جلو رانده شود.

معماری ذخیرهسازی در ClickHouse: موتور ReplacingMergeTree

کلیکهاوس برای درجهای تکردیفی (Single-Row Inserts) یا دستورات UPDATE/DELETE لحظهای طراحی نشده است؛ زیرا هر بار درج، یک پوشه فیزیکی جدید (Data Part) روی دیسک میسازد و درج تکردیفی باعث بروز خطای معروف Too many parts میشود. برای پیادهسازی CDC روی کلیکهاوس، از موتور ReplacingMergeTree به همراه دو ستون کنترلی استفاده میکنیم:

• ستون نسخه (`_version`): معمولاً مقدار ts_ms (زمان تراکنش در منبع) یا lsn در نظر گرفته میشود تا کلیکهاوس هنگام ادغام پسزمینه (Background Merge)، تنها جدیدترین نسخه هر کلید اصلی (ORDER BY) را نگه دارد.
• ستون حذف منطقی (`_is_deleted`): یک فیلد UInt8 که در صورت دریافت رویداد حذف (op == 'd') از Debezium، مقدار 1 میگیرد و رکورد را در فرآیند Merge پاکسازی میکند.

پیادهسازی Consumer میکرو-بچ پایتون از Kafka به ClickHouse

در کد زیر، یک سرویس مصرفکننده مقاوم در برابر خطا پیادهسازی شده است که رویدادهای CDC تولیدشده توسط Debezium را به صورت دستههای ۵۰۰۰ تایی (یا بازه زمانی ۲ ثانیهای) از Kafka جمعآوری کرده، عملیاتهای ایجاد، ویرایش و حذف را به فرمت ستونی تبدیل میکند و به صورت اتمیک در جدول ReplacingMergeTree کلیکهاوس درج مینماید:

import json
import time
from typing import List, Dict, Any
from confluent_kafka import Consumer, KafkaException
import clickhouse_connect

CLICKHOUSE_DDL = """
CREATE TABLE IF NOT EXISTS analytics.orders_cdc
(
    order_id UInt64,
    user_id UInt64,
    status LowCardinality(String),
    total_amount Decimal(12, 2),
    updated_at DateTime64(3, 'UTC'),
    _version UInt64,
    _is_deleted UInt8
)
ENGINE = ReplacingMergeTree(_version, _is_deleted)
PARTITION BY toYYYYMM(updated_at)
ORDER BY (order_id);
"""

class DebeziumClickHouseSink:
    def __init__(self, kafka_conf: Dict[str, Any], ch_host: str, batch_size: int = 5000, flush_interval: float = 2.0):
        self.consumer = Consumer(kafka_conf)
        self.ch_client = clickhouse_connect.get_client(host=ch_host, compress='lz4')
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.ch_client.command(CLICKHOUSE_DDL)

    @staticmethod
    def _transform_debezium_event(payload: Dict[str, Any]) -> List[Any]:
        op = payload.get('op')  # 'c': create, 'u': update, 'd': delete, 'r': snapshot read
        ts_ms = int(payload.get('ts_ms', int(time.time() * 1000)))
        is_deleted = 1 if op == 'd' else 0
        record = payload.get('before') if op == 'd' else payload.get('after')

        return [
            int(record['order_id']),
            int(record['user_id']),
            str(record.get('status', 'UNKNOWN')),
            float(record.get('total_amount', 0.0)),
            int(record.get('updated_at', ts_ms)),
            ts_ms,
            is_deleted
        ]

    def run(self, topic: str) -> None:
        self.consumer.subscribe([topic])
        buffer: List[List[Any]] = []
        last_flush = time.monotonic()

        try:
            while True:
                msg = self.consumer.poll(timeout=0.2)
                if msg is not None and not msg.error():
                    raw_value = msg.value()
                    if raw_value:  # Skip Kafka tombstone records (null payload)
                        envelope = json.loads(raw_value.decode('utf-8'))
                        payload = envelope.get('payload', envelope)
                        if payload and payload.get('op') in ('c', 'u', 'd', 'r'):
                            buffer.append(self._transform_debezium_event(payload))

                now = time.monotonic()
                if len(buffer) >= self.batch_size or (buffer and (now - last_flush) >= self.flush_interval):
                    self.ch_client.insert(
                        table='analytics.orders_cdc',
                        data=buffer,
                        column_names=['order_id', 'user_id', 'status', 'total_amount', 'updated_at', '_version', '_is_deleted']
                    )
                    # Commit Kafka offsets only after ClickHouse confirms batch persistence
                    self.consumer.commit(asynchronous=False)
                    buffer.clear()
                    last_flush = now
        finally:
            self.consumer.close()