ساختن یک برنامه خط لوله داده بلادرنگ با ابزارهای Change Data Capture: Debezium، Kafka و NiFi

Summarize this content to 400 words in Persian Lang
Change Data Capture (CDC) به یک تکنیک حیاتی برای یکپارچه سازی داده های مدرن تبدیل شده است که به سازمان ها اجازه می دهد تغییرات داده ها را در سیستم های مختلف در زمان واقعی ردیابی و انتشار دهند. در این مقاله، نحوه ساخت یک راه حل جامع CDC با استفاده از ابزارهای منبع باز قدرتمند مانند Debezium، Apache Kafka و Apache NiFi را بررسی خواهیم کرد.

فن آوری های کلیدی در پشته CDC ما

Debezium: یک پلتفرم منبع باز برای تغییر داده ها که از چندین منبع پایگاه داده پشتیبانی می کند.
آپاچی کافکا: یک پلت فرم پخش توزیع شده که به عنوان سیستم عصبی مرکزی برای خط لوله داده ما عمل می کند.
Apache NiFi: ابزاری برای مدیریت جریان داده که به ما کمک می کند جریان های داده را مسیریابی، تبدیل و پردازش کنیم.

نمای کلی معماریمعماری پیشنهادی ما این مراحل کلیدی را دنبال می کند:

با استفاده از Debezium تغییرات پایگاه داده را ضبط کنید
جریان از طریق کافکا تغییر می کند
پردازش و مسیریابی داده ها با استفاده از NiFi
ذخیره یا پردازش بیشتر داده های تبدیل شده

نمونه رویکرد پیاده سازی

from confluent_kafka import Consumer, Producer
import json
import debezium

class CDCDataPipeline:
def __init__(self, source_db, kafka_bootstrap_servers):
“””
Initialize CDC pipeline with database source and Kafka configuration

:param source_db: Source database connection details
:param kafka_bootstrap_servers: Kafka broker addresses
“””
self.source_db = source_db
self.kafka_servers = kafka_bootstrap_servers

# Debezium connector configuration
self.debezium_config = {
‘connector.class’: ‘io.debezium.connector.mysql.MySqlConnector’,
‘tasks.max’: ‘1’,
‘database.hostname’: source_db[‘host’],
‘database.port’: source_db[‘port’],
‘database.user’: source_db[‘username’],
‘database.password’: source_db[‘password’],
‘database.server.name’: ‘my-source-database’,
‘database.include.list’: source_db[‘database’] }

def start_capture(self):
“””
Start change data capture process
“””
# Configure Kafka producer for streaming changes
producer = Producer({
‘bootstrap.servers’: self.kafka_servers,
‘client.id’: ‘cdc-change-producer’
})

# Set up Debezium connector
def handle_record(record):
“””
Process each captured change record
“””
# Transform record and publish to Kafka
change_event = {
‘source’: record.source(),
‘operation’: record.operation(),
‘data’: record.after()
}

producer.produce(
topic=”database-changes”,
value=json.dumps(change_event)
)

# Start Debezium connector
debezium.start_connector(
config=self.debezium_config,
record_handler=handle_record
)

# Example usage
source_database = {
‘host’: ‘localhost’,
‘port’: 3306,
‘username’: ‘cdc_user’,
‘password’: ‘secure_password’,
‘database’: ‘customer_db’
}

pipeline = CDCDataPipeline(
source_database,
kafka_bootstrap_servers=”localhost:9092″
)
pipeline.start_capture()

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

مراحل اجرای دقیق

پیکربندی منبع پایگاه داده اولین مرحله شامل پیکربندی Debezium برای اتصال به پایگاه داده منبع شما است. این مستلزم:

مجوزهای کاربر پایگاه داده مناسب
اتصال به شبکه
فعال کردن ثبت باینری (برای MySQL)

کافکا به عنوان یک پلتفرم جریان آپاچی کافکا به عنوان یک واسطه پیام مرکزی عمل می کند و رویدادهای تغییر را ثبت و ذخیره می کند. ملاحظات کلیدی عبارتند از:

پیکربندی پارتیشن های موضوعی
تنظیم سیاست های نگهداری مناسب
پیاده سازی معنایی پردازش دقیقاً یک بار

تبدیل داده با NiFi Apache NiFi قابلیت های قدرتمند مسیریابی و تبدیل داده را ارائه می دهد:

رویدادهای تغییر مسیر و فیلتر
غنی سازی داده ها را اعمال کنید
منطق تحول پیچیده را مدیریت کنید

چالش ها و بهترین شیوه ها

مدیریت تغییرات طرحواره: استراتژی های قوی تکامل طرحواره را اجرا کنید
بهینه سازی عملکرد: از پارتیشن بندی و فشرده سازی مناسب استفاده کنید
مدیریت خطا: مکانیزم های جامع ردیابی خطا و تلاش مجدد را پیاده سازی کنید

مخزن GitHub

من یک نمونه پیاده سازی ایجاد کرده ام که می توانید آن را بررسی کرده و به عنوان مرجع از آن استفاده کنید. کد کامل و اسناد اضافی را می توانید در آدرس زیر بیابید:مخزن GitHub: https://github.com/Angelica-R/cdc-data-pipeline

نتیجه گیریایجاد راه حل Change Data Capture نیازمند طراحی دقیق معماری و انتخاب ابزار مناسب است. با استفاده از Debezium، Kafka و NiFi، می‌توانید یک پلتفرم یکپارچه‌سازی داده‌های مقیاس‌پذیر و قوی ایجاد کنید که بینش‌های بی‌درنگ درباره تغییرات داده‌های شما ارائه می‌دهد.

from confluent_kafka import Consumer, Producer
import json
import debezium

class CDCDataPipeline:
    def __init__(self, source_db, kafka_bootstrap_servers):
        """
        Initialize CDC pipeline with database source and Kafka configuration

        :param source_db: Source database connection details
        :param kafka_bootstrap_servers: Kafka broker addresses
        """
        self.source_db = source_db
        self.kafka_servers = kafka_bootstrap_servers

        # Debezium connector configuration
        self.debezium_config = {
            'connector.class': 'io.debezium.connector.mysql.MySqlConnector',
            'tasks.max': '1',
            'database.hostname': source_db['host'],
            'database.port': source_db['port'],
            'database.user': source_db['username'],
            'database.password': source_db['password'],
            'database.server.name': 'my-source-database',
            'database.include.list': source_db['database']
        }

    def start_capture(self):
        """
        Start change data capture process
        """
        # Configure Kafka producer for streaming changes
        producer = Producer({
            'bootstrap.servers': self.kafka_servers,
            'client.id': 'cdc-change-producer'
        })

        # Set up Debezium connector
        def handle_record(record):
            """
            Process each captured change record
            """
            # Transform record and publish to Kafka
            change_event = {
                'source': record.source(),
                'operation': record.operation(),
                'data': record.after()
            }

            producer.produce(
                topic="database-changes", 
                value=json.dumps(change_event)
            )

        # Start Debezium connector
        debezium.start_connector(
            config=self.debezium_config,
            record_handler=handle_record
        )

# Example usage
source_database = {
    'host': 'localhost',
    'port': 3306,
    'username': 'cdc_user',
    'password': 'secure_password',
    'database': 'customer_db'
}

pipeline = CDCDataPipeline(
    source_database, 
    kafka_bootstrap_servers="localhost:9092"
)
pipeline.start_capture()