آیا جایگزینی برای Debezium + Kafka وجود دارد؟

Summarize this content to 400 words in Persian Lang
مدتی پیش این سوال را در Reddit پرسیدم و پاسخ های ارزشمند زیادی دریافت کردم.

بنابراین، من به هر پاسخ نگاه کردم و نتایج را در این مقاله مستند کردم.

TL; DR

نه، دبیزیوم علیرغم برخی اشکالات در حال حاضر بر بازار تسلط دارد.

توضیح پیشینه

چرا می خواهیم جایگزینی برای Debezium پیدا کنیم؟ دلیل اصلی این است که ما با یک سناریوی چالش برانگیز مواجه شدیم.

این یک سناریوی معمولی برای Debezium است، که در آن هرگونه تغییر در منبع داده ثبت شده و برای پردازش پایین دست به کافکا داده می‌شود.

مزیت این معماری ساده و کارآمد است و تضمین می کند که تمام فرآیندهای پایین دست تا حد امکان بلادرنگ هستند.

اگر منبع دارای تعداد زیادی به روز رسانی باشد، Debezium می تواند به صورت افقی مقیاس شود تا زمانی که تعداد زیادی از به روز رسانی ها در یک جدول متمرکز شوند. اینجاست که Debezium به محدودیت های خود می رسد.

حتی اگر Debezium می تواند به صورت افقی مقیاس شود، به این معنی است که به روز رسانی های اولیه توسط یک فرآیند می توانند در چندین فرآیند توزیع شوند. اگر هر جدول قبلاً یک فرآیند اختصاصی داشته باشد، مقیاس بندی افقی دیگر امکان پذیر نیست.

ما در چنین شرایطی قرار داریم، در محیط خود، حتی اگر مشخصات دستگاه کشیده شده باشد، توان خروجی CDC یک جدول تنها 25 مگابایت بر ثانیه است.

مطمئناً این یک مورد معمولی نیست، به هر حال، تغییر 25 مگابایت بر ثانیه برای یک جدول کاملاً قابل توجه است. با این حال، اگر با منبع داده ای مواجه شویم که در حال انجام مهاجرت داده در مقیاس بزرگ است، این محدودیت به راحتی قابل نقض است.

برای اطمینان از عملکرد بلادرنگ خط لوله داده خود در پایین دست، فقط می‌توانیم از بالادست بخواهیم هنگام مواجهه با این سطح از مهاجرت داده، مهربان باشد و سعی کند کار محدود کردن سرعت را به خوبی انجام دهد.

با این حال، این محدودیت بهره وری توسعه دهندگان بالادستی را تا حد زیادی کاهش می دهد. از یک طرف، آنها باید فرآیند حسابرسی را به تعمیر و نگهداری عادی خود اضافه کنند، و از طرف دیگر، باید یک محدودیت نرخ اضافی برای هر تعمیر و نگهداری ایجاد کنند.

پس بیایید راه حلی پیدا کنیم.

بررسی اجمالی راه حل

راه حل های زیر از آن مقاله Reddit جمع آوری شده است.

جریان مصب
جریان
Fivetran HVR
CDC اختصاصی
مجرا

سه راه حل اول خدمات سازمانی بدون منبع باز هستند، بنابراین برای ما کار نمی کنند. از این گذشته، ما در تلاش هستیم تا یک مورد استفاده جزئی را حل کنیم، نه یک انجام کامل.

اگرچه Estuary Flow می گوید که آنها یک رویکرد استقرار محلی دارند، من نتوانستم اطلاعاتی در مورد آن پیدا کنم.

راه حل چهارم این بود که خودمان ابزار جدیدی را توسعه دهیم که به اعتقاد من راه حل اساسی برای مشکل خواهد بود. به هر حال، Debezium در JAVA توسعه یافته است و ما باید بتوانیم با Golang، Rust یا حتی C/C++ به عملکرد بهتری دست یابیم. با این حال، هزینه توسعه برای ما بسیار بالا بود و شروع از صفر دشوار بود.

چهار گزینه اول نیازهای ما را برآورده نکرد، اما گزینه پنجم به عنوان یک راه حل امیدوارکننده توجه من را به خود جلب کرد.

Conduit یک پلت فرم انتقال داده منبع باز است که در Golang توسعه یافته است و انواع اتصال دهنده ها را برای ادغام بسیاری از فروشگاه های داده فراهم می کند. علاوه بر این، ما همچنین می توانیم مبدل خود را برای انجام پیش پردازش فرمت داده توسعه دهیم.

بنابراین، من شروع به آزمایش عملکرد Conduit کردم.

محیط انقضا

برای ساده نگه داشتن همه چیز، من از Kafka Connect به جای Debezium استفاده کردم. این دو اساسا یکسان هستند، اما با توزیع کننده های متفاوت، و در پشت صحنه همه آنها از یک کتابخانه استفاده می کنند.

Locust مسئول ایجاد تغییرات MongoDB است، سپس Conduit و Kafka Connect روی موضوعات مختلف کافکا می نویسند.

ما می توانیم سرعت نوشتن موضوعات کافکا را مشاهده کنیم تا مشخص کنیم چه کسی عملکرد بهتری دارد.

کل محیط آزمایش به شرح زیر است.

https://gist.github.com/wirelessr/82a642685d40d78a49a4cdb1ff1cfa9f

من از دو تصویر بسته بندی شده خودم، Conduit و Kafka Connect استفاده می کنم که دارای اتصال MongoDB هستند.

ایجاد مقدار زیادی تغییر با پر کردن MongoDB با یک دسته اسناد چرب و سپس تغییر مقدار یک فیلد در همه اسناد آسان است.

اسکریپت ملخ استفاده شده به شرح زیر است.

import time

from locust import User, task, between
from faker import Faker
import pymongo

fake = Faker()

class MongoDBUser(User):
wait_time = between(1, 5)

def __init__(self, environment):
super().__init__(environment)

self.env = environment

def on_start(self):
self.client = pymongo.MongoClient(self.host)
self.db = self.client[“test”] self.collection = self.db[“test_new”]

@task
def incr_seq(self):
response = None
exception = None
start_perf_counter = time.perf_counter()
response_length = 0

try:
response = self.collection.update_many(
{},
{“$inc”: {“seq”: 1}}
)
response_length = response.matched_count
except Exception as e:
exception = e

self.env.events.request.fire(
request_type=”mongo”,
name=”incr seq”,
response_time=(time.perf_counter() – start_perf_counter) * 1000,
response_length=response_length,
response=response,
context=None,
exception=exception,
)

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

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

بارگذاری نتیجه آزمایش

برای این آزمایش، من از یک ماشین محلی استفاده کردم بدون اینکه به طور کامل به CPU یا حافظه آن فشار وارد کنم، و برخی منابع را برای جلوگیری از خطاهای ناشی از گلوگاه های عملکرد در دسترس گذاشتم.

به عبارت دیگر، این تست توانایی منظم یک فرآیند واحد را برای رسیدگی به یک بار جدول نشان می دهد.

مجرا

کافکا کانکت

همانطور که نتایج نشان می‌دهد، زمانی که منابع سیستم کافی باشد، توان عملیاتی کافکا به طور قابل توجهی بهتر از Conduit است.

من در مورد این نتیجه کمی گیج شده بودم، بنابراین تست را چند بار تکرار کردم، اما اعداد مشابهی به دست آوردم.

بسته بندی کنید

بازگشت به سوال در عنوان.

آیا جایگزینی برای Debezium + Kafka وجود دارد؟

در حال حاضر نه – حداقل، نه در میان ابزارهای منبع باز.

من در Reddit پرسیده ام، اما شاید Medium پاسخ متفاوتی داشته باشد، بنابراین با خیال راحت راه حل های خود را ارائه دهید.

import time

from locust import User, task, between
from faker import Faker
import pymongo

fake = Faker()

class MongoDBUser(User):
    wait_time = between(1, 5)

    def __init__(self, environment):
        super().__init__(environment)

        self.env = environment

    def on_start(self):
        self.client = pymongo.MongoClient(self.host)
        self.db = self.client["test"]
        self.collection = self.db["test_new"]

    @task
    def incr_seq(self):
        response = None
        exception = None
        start_perf_counter = time.perf_counter()
        response_length = 0

        try:
            response = self.collection.update_many(
                {},
                {"$inc": {"seq": 1}}
            )
            response_length = response.matched_count
        except Exception as e:
            exception = e

        self.env.events.request.fire(
            request_type="mongo",
            name="incr seq",
            response_time=(time.perf_counter() - start_perf_counter) * 1000,
            response_length=response_length,
            response=response,
            context=None,
            exception=exception,
        )