فصل ۱۱ — پردازش جریانی

یک سیستم پیچیده که کار می‌کند، همیشه از یک سیستم سادهٔ کارآمد تکامل یافته است. گزارهٔ معکوس هم به نظر درست می‌آید: سیستم پیچیده‌ای که از صفر طراحی شده هرگز کار نمی‌کند و نمی‌توان آن را وادار به کار کرد. — John Gall، Systemantics (۱۹۷۵)

در فصل ۱۰ دربارهٔ پردازش دسته‌ای — تکنیک‌هایی که مجموعه‌ای از فایل‌ها را به‌عنوان ورودی می‌خوانند و مجموعه‌ای جدید از فایل‌های خروجی تولید می‌کنند — بحث کردیم. خروجی شکلی از دادهٔ مشتق‌شده است؛ یعنی مجموعه‌داده‌ای که در صورت لزوم می‌توان با اجرای دوبارهٔ فرایند دسته‌ای بازسازی کرد. دیدیم این ایدهٔ ساده اما قدرتمند چگونه می‌تواند برای ساخت ایندکس جستجو، سیستم‌های توصیه، تحلیل و موارد دیگر به کار رود.

با این حال، در سراسر فصل ۱۰ یک فرض بزرگ باقی ماند: ورودی محدود است — یعنی اندازه‌ای شناخته‌شده و متناهی دارد — بنابراین فرایند دسته‌ای می‌داند چه زمانی خواندن ورودی تمام شده. مثلاً عملیات مرتب‌سازی که در MapReduce محوری است باید کل ورودی را بخواند تا بتواند خروجی تولید کند: ممکن است آخرین رکورد ورودی کمترین کلید را داشته باشد و بنابراین باید اولین رکورد خروجی باشد؛ پس شروع زودهنگام خروجی گزینه نیست.

در واقعیت، بسیاری از داده‌ها نامحدودند چون به‌تدریج در طول زمان می‌رسند: کاربران دیروز و امروز داده تولید کرده‌اند و فردا هم ادامه می‌دهند. مگر اینکه کسب‌وکارتان ورشکست شود، این فرایند هرگز پایان نمی‌یابد و مجموعه‌داده هرگز به معنای مفید «کامل» نمی‌شود [۱]. بنابراین پردازنده‌های دسته‌ای باید مصنوعی داده را به قطعه‌های با مدت ثابت تقسیم کنند: مثلاً پردازش یک روز داده در پایان هر روز، یا یک ساعت داده در پایان هر ساعت.

مشکل فرایندهای دسته‌ای روزانه این است که تغییرات ورودی فقط یک روز بعد در خروجی منعکس می‌شوند، که برای بسیاری از کاربران بی‌صبر خیلی کند است. برای کاهش تأخیر می‌توانیم پردازش را مکررتر اجرا کنیم — مثلاً یک ثانیه داده در پایان هر ثانیه — یا حتی پیوسته، و برش‌های زمانی ثابت را کنار بگذاریم و هر رویداد را همان لحظهٔ وقوع پردازش کنیم. این ایدهٔ پشت پردازش جریانی است.

به‌طور کلی «جریان» (stream) به داده‌ای اشاره دارد که به‌تدریج در طول زمان در دسترس قرار می‌گیرد. این مفهوم در جاهای زیادی ظاهر می‌شود: در stdin و stdout یونیکس، زبان‌های برنامه‌نویسی (لیست‌های lazy) [۲]، APIهای فایل‌سیستم (مثل FileInputStream جاوا)، اتصال‌های TCP، تحویل صوت و ویدئو روی اینترنت و غیره.

در این فصل جریان‌های رویداد را به‌عنوان مکانیزم مدیریت داده بررسی می‌کنیم: همتای پردازش تدریجی و نامحدود دادهٔ دسته‌ای فصل قبل. ابتدا بحث می‌کنیم جریان‌ها چگونه نمایش داده، ذخیره و از شبکه منتقل می‌شوند. در «پایگاه‌های داده و جریان‌ها» در صفحه ۴۵۱ رابطهٔ جریان‌ها و پایگاه‌های داده را بررسی می‌کنیم. و در نهایت در «پردازش جریان‌ها» در صفحه ۴۶۴ رویکردها و ابزارهای پردازش پیوستهٔ آن جریان‌ها و کاربردشان در ساخت برنامه‌ها را می‌بینیم.

انتقال جریان‌های رویداد

در دنیای پردازش دسته‌ای، ورودی و خروجی یک job فایل است (شاید روی فایل‌سیستم توزیع‌شده). معادل جریانی آن چه شکلی است؟

وقتی ورودی فایل است (توالی بایت‌ها)، اولین گام پردازش معمولاً parse کردن آن به توالی رکوردهاست. در زمینهٔ پردازش جریانی، رکورد بیشتر رویداد (event) نامیده می‌شود، اما اساساً همان است: شیء کوچک، خودکفا و تغییرناپذیر حاوی جزئیات چیزی که در زمانی رخ داده. رویداد معمولاً timestamp دارد که زمان وقوع را طبق ساعت time-of-day نشان می‌دهد (به «ساعت‌های monotonic در برابر time-of-day» در صفحه ۲۸۸ مراجعه کنید).

مثلاً آنچه رخ داده ممکن است عملی کاربر باشد، مثل مشاهدهٔ صفحه یا خرید. ممکن است از ماشین بیاید، مثل اندازه‌گیری دوره‌ای از سنسور دما یا متریک استفاده از CPU. در مثال «پردازش دسته‌ای با ابزارهای یونیکس» در صفحه ۳۹۱، هر خط لاگ وب‌سرور یک رویداد است.

رویداد ممکن است به‌صورت رشتهٔ متنی، JSON یا شکل دودویی کدگذاری شود، همان‌طور که در فصل ۴ بحث شد. این کدگذاری اجازه می‌دهد رویداد را ذخیره کنید — مثلاً با append به فایل، insert در جدول رابطه‌ای یا نوشتن در document database. همچنین اجازه می‌دهد رویداد را از شبکه به گره دیگر برای پردازش بفرستید.

در پردازش دسته‌ای، فایل یک‌بار نوشته می‌شود و سپس شاید چند job آن را بخوانند. به‌طور مشابه، در اصطلاحات جریانی، رویداد یک‌بار توسط تولیدکننده (producer؛ publisher یا sender) تولید می‌شود و سپس شاید چند مصرف‌کننده (consumer؛ subscriber یا recipient) آن را پردازش کنند [۳]. در فایل‌سیستم، نام فایل مجموعه‌ای از رکوردهای مرتبط را مشخص می‌کند؛ در سیستم جریانی، رویدادهای مرتبط معمولاً در یک topic یا stream گروه‌بندی می‌شوند.

در اصل، فایل یا پایگاه داده برای اتصال تولیدکنندگان و مصرف‌کنندگان کافی است: تولیدکننده هر رویدادی که تولید می‌کند در datastore می‌نویسد و هر مصرف‌کننده دوره‌ای datastore را poll می‌کند تا رویدادهای جدید از آخرین اجرا را ببیند. این اساساً همان کاری است که فرایند دسته‌ای هنگام پردازش یک روز داده در پایان هر روز انجام می‌دهد.

با این حال، وقتی به سمت پردازش پیوسته با تأخیر کم می‌روید، polling اگر datastore برای این کار طراحی نشده باشد گران می‌شود. هرچه بیشتر poll کنید، درصد درخواست‌هایی که رویداد جدید برمی‌گردانند کمتر و سربار بیشتر می‌شود. بهتر است مصرف‌کنندگان هنگام ظاهر شدن رویداد جدید مطلع شوند.

پایگاه‌های داده سنتی این مکانیزم اعلان را خوب پشتیبانی نکرده‌اند: پایگاه‌های رابطه‌ای معمولاً trigger دارند که به تغییر (مثلاً insert ردیف در جدول) واکنش نشان می‌دهند، اما کارشان محدود است و تا حدی پس‌اندیش در طراحی پایگاه داده بوده‌اند [۴، ۵]. در عوض ابزارهای تخصصی برای تحویل اعلان رویداد توسعه یافته‌اند.

سیستم‌های پیام‌رسانی

رویکرد رایج برای اطلاع‌رسانی به مصرف‌کنندگان دربارهٔ رویدادهای جدید، استفاده از سیستم پیام‌رسانی است: تولیدکننده پیامی حاوی رویداد می‌فرستد که به مصرف‌کنندگان push می‌شود. قبلاً در «جریان داده با ارسال پیام» در صفحه ۱۳۶ به این سیستم‌ها اشاره کردیم، اما اکنون جزئیات بیشتری می‌بینیم.

کانال ارتباطی مستقیم مثل pipe یونیکس یا اتصال TCP بین تولیدکننده و مصرف‌کننده راه ساده‌ای برای پیاده‌سازی سیستم پیام‌رسانی است. با این حال، بیشتر سیستم‌های پیام‌رسانی این مدل پایه را گسترش می‌دهند. به‌ویژه pipe و TCP یونیکس دقیقاً یک فرستنده را به یک گیرنده وصل می‌کنند، در حالی که سیستم پیام‌رسانی به چند تولیدکننده اجازه می‌دهد به یک topic پیام بفرستند و به چند مصرف‌کننده اجازه می‌دهد پیام‌های یک topic را دریافت کنند.

در این مدل publish/subscribe، سیستم‌های مختلف رویکردهای گسترده‌ای دارند و پاسخ واحدی برای همهٔ کاربردها نیست. برای تمایز، پرسیدن این دو سؤال مفید است:

۱. اگر تولیدکنندگان سریع‌تر از توان پردازش مصرف‌کنندگان پیام بفرستند چه می‌شود؟ به‌طور کلی سه گزینه وجود دارد: سیستم پیام‌ها را drop کند، در صف buffer کند، یا backpressure (کنترل جریان؛ یعنی مسدود کردن تولیدکننده از ارسال پیام بیشتر) اعمال کند. مثلاً pipe و TCP یونیکس از backpressure استفاده می‌کنند: buffer ثابت و کوچکی دارند و اگر پر شود، فرستنده تا زمانی که گیرنده داده را از buffer خارج کند مسدود می‌شود (به «ازدحام شبکه و صف‌بندی» در صفحه ۲۸۲ مراجعه کنید).

اگر پیام‌ها در صف buffer شوند، مهم است بفهمید با رشد صف چه می‌شود. آیا سیستم اگر صف دیگر در حافظه جا نشود crash می‌کند، یا پیام‌ها را روی دیسک می‌نویسد؟ در آن صورت دسترسی دیسک چه اثری بر کارایی سیستم پیام‌رسانی دارد [۶]؟

۲. اگر گره‌ها crash کنند یا موقتاً offline شوند — آیا پیامی گم می‌شود؟ مانند پایگاه‌های داده، durability ممکن است ترکیبی از نوشتن روی دیسک و/یا replication نیاز داشته باشد (به کادر کناری «رپلیکاسیون و durability» در صفحه ۲۲۷ مراجعه کنید)، که هزینه دارد. اگر گاهی از دست دادن پیام قابل قبول است، احتمالاً throughput بالاتر و latency پایین‌تری روی همان سخت‌افزار می‌گیرید.

قابل‌قبول بودن از دست دادن پیام به‌طور زیاد به کاربرد بستگی دارد. مثلاً در خوانش سنسور و متریک‌های دوره‌ای، گاهی گم شدن یک نقطهٔ داده شاید مهم نباشد چون مقدار به‌روز شده به‌زودی ارسال می‌شود. با این حال، اگر تعداد زیادی پیام drop شود، ممکن است فوراً مشخص نباشد متریک‌ها نادرست‌اند [۷]. اگر رویدادها را می‌شمارید، تحویل قابل‌اعتماد مهم‌تر است چون هر پیام گم‌شده یعنی شمارندهٔ نادرست.

ویژگی خوب سیستم‌های پردازش دسته‌ای فصل ۱۰ تضمین قوی قابلیت اطمینان است: taskهای شکست‌خورده خودکار retry می‌شوند و خروجی جزئی taskهای شکست‌خورده خودکار دور ریخته می‌شود. یعنی خروجی همان‌طور است که گویی هیچ خرابی رخ نداده، که مدل برنامه‌نویسی را ساده می‌کند. بعد در این فصل می‌بینیم چگونه تضمین‌های مشابه در زمینهٔ جریانی فراهم کنیم.

پیام‌رسانی مستقیم از تولیدکنندگان به مصرف‌کنندگان

تعدادی سیستم پیام‌رسانی از ارتباط شبکهٔ مستقیم بین تولیدکنندگان و مصرف‌کنندگان بدون عبور از گره‌های واسطه استفاده می‌کنند:

• UDP multicast در صنعت مالی برای جریان‌هایی مثل feed بازار سهام، که latency پایین مهم است، به‌طور گسترده به کار می‌رود [۸]. اگرچه خود UDP غیرقابل‌اعتماد است، پروتکل‌های سطح کاربرد می‌توانند بسته‌های گم‌شده را بازیابی کنند (تولیدکننده باید بسته‌های ارسال‌شده را به‌خاطر بسپارد تا در صورت نیاز retransmit کند).
• کتابخانه‌های پیام‌رسانی بدون broker مثل ZeroMQ [۹] و nanomsg رویکرد مشابهی دارند و publish/subscribe را روی TCP یا IP multicast پیاده می‌کنند.
• StatsD [۱۰] و Brubeck [۷] از پیام‌رسانی UDP غیرقابل‌اعتماد برای جمع‌آوری متریک از همهٔ ماشین‌های شبکه و مانیتورینگ استفاده می‌کنند. (در پروتکل StatsD، شمارنده‌های counter فقط وقتی درست‌اند که همهٔ پیام‌ها دریافت شوند؛ استفاده از UDP متریک‌ها را در بهترین حالت تقریبی می‌کند [۱۱]. به «TCP در برابر UDP» در صفحه ۲۸۳ هم مراجعه کنید.)
• اگر مصرف‌کننده سرویسی روی شبکه expose کند، تولیدکنندگان می‌توانند درخواست HTTP یا RPC مستقیم (به «جریان داده از طریق سرویس‌ها: REST و RPC» در صفحه ۱۳۱ مراجعه کنید) برای push پیام به مصرف‌کننده بزنند. این ایدهٔ webhooks [۱۲] است: الگویی که URL callback یک سرویس در سرویس دیگر ثبت می‌شود و هنگام وقوع رویداد به آن URL درخواست می‌زند.

اگرچه این سیستم‌های پیام‌رسانی مستقیم برای موارد طراحی‌شده خوب کار می‌کنند، معمولاً کد برنامه باید احتمال از دست رفتن پیام را در نظر بگیرد. خرابی‌هایی که تحمل می‌کنند محدودند: حتی اگر پروتکل‌ها بسته‌های گم‌شده در شبکه را تشخیص و retransmit کنند، معمولاً فرض می‌کنند تولیدکننده و مصرف‌کننده همیشه online‌اند.

اگر مصرف‌کننده offline باشد، ممکن است پیام‌هایی را که در زمان غیرقابل‌دسترس بودن ارسال شده‌اند از دست بدهد. برخی پروتکل‌ها به تولیدکننده اجازه می‌دهند تحویل ناموفق را retry کنند، اما اگر تولیدکننده crash کند و buffer پیام‌هایی که باید retry شوند را از دست بدهد، این رویکرد شکست می‌خورد.

کارگزارهای پیام

جایگزین پرکاربرد، ارسال پیام از طریق message broker (message queue) است که اساساً نوعی پایگاه داده بهینه‌شده برای مدیریت جریان پیام است [۱۳]. به‌عنوان سرور اجرا می‌شود و تولیدکنندگان و مصرف‌کنندگان به‌عنوان کلاینت متصل می‌شوند. تولیدکنندگان پیام می‌نویسند و مصرف‌کنندگان با خواندن از broker دریافت می‌کنند.

با متمرکز کردن داده در broker، این سیستم‌ها راحت‌تر کلاینت‌هایی که می‌آیند و می‌روند (اتصال، قطع، crash) را تحمل می‌کنند و مسئلهٔ durability به broker منتقل می‌شود. برخی brokerها پیام را فقط در حافظه نگه می‌دارند، برخی دیگر (بسته به پیکربندی) روی دیسک می‌نویسند تا در crash broker از بین نرود. در برابر مصرف‌کنندگان کند، معمولاً صف‌بندی نامحدود اجازه می‌دهند (به‌جای drop یا backpressure)، هرچند این هم ممکن است به پیکربندی بستگی داشته باشد.

پیامد صف‌بندی این است که مصرف‌کنندگان معمولاً ناهمگام‌اند: وقتی تولیدکننده پیام می‌فرستد، معمولاً فقط منتظر تأیید buffer شدن پیام توسط broker می‌ماند و منتظر پردازش توسط مصرف‌کنندگان نمی‌ماند. تحویل به مصرف‌کنندگان در زمانی نامشخص آینده رخ می‌دهد — اغلب در کسری از ثانیه، اما گاهی با تأخیر قابل‌توجه اگر backlog صف وجود داشته باشد.

کارگزارهای پیام در مقایسه با پایگاه‌های داده

برخی message brokerها حتی می‌توانند در پروتکل‌های two-phase commit با XA یا JTA شرکت کنند (به «تراکنش‌های توزیع‌شده در عمل» در صفحه ۳۶۰ مراجعه کنید). این ویژگی آن‌ها را از نظر ماهیت به پایگاه‌های داده نزدیک می‌کند، اگرچه تفاوت‌های عملی مهم بین broker و پایگاه داده باقی می‌ماند:

• پایگاه‌های داده معمولاً داده را تا حذف صریح نگه می‌دارند، در حالی که بیشتر message brokerها پس از تحویل موفق به مصرف‌کنندگان پیام را خودکار حذف می‌کنند. چنین brokerهایی برای ذخیرهٔ بلندمدت داده مناسب نیستند.
• چون پیام‌ها را سریع حذف می‌کنند، بیشتر brokerها فرض می‌کنند working set نسبتاً کوچک است — یعنی صف‌ها کوتاه‌اند. اگر broker به‌خاطر کندی مصرف‌کنندگان باید پیام زیادی buffer کند (شاید با spill به دیسک اگر در حافظه جا نشود)، پردازش هر پیام طولانی‌تر و throughput کلی ممکن است افت کند [۶].
• پایگاه‌های داده اغلب secondary index و روش‌های مختلف جستجوی داده دارند، در حالی که message brokerها اغلب روشی برای subscribe به زیرمجموعه‌ای از topicها با الگویی مشخص پشتیبانی می‌کنند. مکانیزم‌ها متفاوت‌اند، اما هر دو راهی برای انتخاب بخشی از داده‌ای است که کلاینت می‌خواهد بداند.
• هنگام query روی پایگاه داده، نتیجه معمولاً بر اساس snapshot لحظه‌ای داده است؛ اگر کلاینت دیگری بعداً چیزی بنویسد که نتیجهٔ query را عوض کند، کلاینت اول متوجه نمی‌شود نتیجهٔ قبلی منسوخ شده (مگر query را تکرار کند یا تغییرات را poll کند). در مقابل، message brokerها query دلخواه پشتیبانی نمی‌کنند، اما هنگام تغییر داده (یعنی در دسترس شدن پیام جدید) به کلاینت‌ها اطلاع می‌دهند.

این نمای سنتی message broker است که در استانداردهایی مثل JMS [۱۴] و AMQP [۱۵] و نرم‌افزارهایی مثل RabbitMQ، ActiveMQ، HornetQ، Qpid، TIBCO Enterprise Message Service، IBM MQ، Azure Service Bus و Google Cloud Pub/Sub [۱۶] تجسم یافته است.

چند مصرف‌کننده

وقتی چند مصرف‌کننده پیام‌های یک topic را می‌خوانند، دو الگوی اصلی پیام‌رسانی به کار می‌رود، همان‌طور که در شکل ۱۱-۱ نشان داده شده:

Load balancing هر پیام به یکی از مصرف‌کنندگان تحویل می‌شود تا مصرف‌کنندگان کار پردازش پیام‌های topic را تقسیم کنند. broker ممکن است پیام‌ها را دلخواه به مصرف‌کنندگان تخصیص دهد. این الگو وقتی مفید است که پردازش پیام‌ها گران باشد و بخواهید با افزودن مصرف‌کننده پردازش را موازی کنید. (در AMQP با داشتن چند کلاینت از یک queue مصرف می‌کنید؛ در JMS shared subscription نامیده می‌شود.)

Fan-out هر پیام به همهٔ مصرف‌کنندگان تحویل می‌شود. fan-out به چند مصرف‌کنندهٔ مستقل اجازه می‌دهد هر کدام به همان پخش پیام «گوش دهند» بدون تأثیر بر یکدیگر — معادل جریانی داشتن چند job دسته‌ای مختلف که همان فایل ورودی را می‌خوانند. (این ویژگی در JMS با topic subscription و در AMQP با exchange binding فراهم می‌شود.)

شکل ۱۱-۱. (الف) load balancing: تقسیم کار مصرف یک topic بین مصرف‌کنندگان؛ (ب) fan-out: تحویل هر پیام به چند مصرف‌کننده.

دو الگو را می‌توان ترکیب کرد: مثلاً دو گروه جدا از مصرف‌کنندگان هر کدام به یک topic subscribe کنند، طوری که هر گروه جمعاً همهٔ پیام‌ها را بگیرد، اما در هر گروه فقط یک گره هر پیام را دریافت کند.

تأییدیه‌ها و تحویل مجدد

مصرف‌کنندگان ممکن است هر لحظه crash کنند؛ بنابراین ممکن است broker پیامی را تحویل دهد اما مصرف‌کننده هرگز آن را پردازش نکند، یا قبل از crash فقط بخشی پردازش کند. برای اطمینان از عدم گم شدن پیام، message brokerها از acknowledgment استفاده می‌کنند: کلاینت باید صریحاً به broker بگوید پردازش پیام تمام شده تا broker آن را از صف حذف کند.

اگر اتصال به کلاینت بدون دریافت acknowledgment بسته یا timeout شود، broker فرض می‌کند پیام پردازش نشده و آن را دوباره به مصرف‌کنندهٔ دیگری تحویل می‌دهد. (توجه: ممکن است پیام واقعاً کاملاً پردازش شده باشد اما acknowledgment در شبکه گم شده باشد. مدیریت این حالت به پروتکل commit اتمی نیاز دارد، همان‌طور که در «تراکنش‌های توزیع‌شده در عمل» در صفحه ۳۶۰ بحث شد.)

همراه با load balancing، رفتار redelivery اثر جالبی روی ترتیب پیام‌ها دارد. در شکل ۱۱-۲، مصرف‌کنندگان معمولاً پیام‌ها را به ترتیب ارسال تولیدکنندگان پردازش می‌کنند. با این حال، مصرف‌کننده ۲ هنگام پردازش پیام m3 crash می‌کند، در حالی که مصرف‌کننده ۱ پیام m4 را پردازش می‌کند. پیام m3 بدون تأیید بعداً دوباره به مصرف‌کننده ۱ تحویل می‌شود، با نتیجهٔ اینکه مصرف‌کننده ۱ پیام‌ها را به ترتیب m4، m3، m5 پردازش می‌کند. بنابراین m3 و m4 به همان ترتیبی که تولیدکننده ۱ فرستاده تحویل نمی‌شوند.

شکل ۱۱-۲. مصرف‌کننده ۲ هنگام پردازش m3 crash می‌کند، بنابراین بعداً دوباره به مصرف‌کننده ۱ تحویل داده می‌شود.

حتی اگر message broker در غیر این صورت سعی کند ترتیب پیام‌ها را حفظ کند (همان‌طور که استانداردهای JMS و AMQP می‌خواهند)، ترکیب load balancing با redelivery ناگزیر به جابه‌جایی ترتیب پیام‌ها منجر می‌شود. برای اجتناب، می‌توانید صف جدا برای هر مصرف‌کننده استفاده کنید (یعنی از load balancing استفاده نکنید). جابه‌جایی ترتیب پیام اگر پیام‌ها کاملاً مستقل باشند مشکل نیست، اما اگر وابستگی علی بین پیام‌ها باشد مهم می‌شود، همان‌طور که بعد در فصل می‌بینیم.

لاگ‌های پارتیشن‌شده

ارسال بسته از شبکه یا درخواست به سرویس شبکه معمولاً عملی گذرا است که اثر دائمی برجا نمی‌گذارد. اگرچه می‌توان آن را دائمی ضبط کرد (با packet capture و logging)، معمولاً این‌طور فکر نمی‌کنیم. حتی message brokerهایی که پیام را پایدار روی دیسک می‌نویسند، پس از تحویل به مصرف‌کنندگان سریع حذف می‌کنند، چون حول ذهنیت پیام‌رسانی گذرا ساخته شده‌اند.

پایگاه‌های داده و فایل‌سیستم‌ها رویکرد معکوس دارند: هر چیزی که به پایگاه داده یا فایل نوشته می‌شود معمولاً انتظار می‌رود دائمی ثبت شود، حداقل تا کسی صریحاً حذف کند.

این تفاوت ذهنی اثر بزرگی روی نحوهٔ ایجاد دادهٔ مشتق‌شده دارد. ویژگی کلیدی فرایندهای دسته‌ای، همان‌طور که در فصل ۱۰ بحث شد، این است که می‌توانید آن‌ها را مکرر اجرا کنید و روی گام‌های پردازش آزمایش کنید بدون خطر آسیب به ورودی (چون ورودی فقط‌خواندنی است). در پیام‌رسانی سبک AMQP/JMS اینطور نیست: دریافت پیام اگر acknowledgment باعث حذف از broker شود مخرب است، پس نمی‌توانید همان مصرف‌کننده را دوباره اجرا کنید و همان نتیجه را انتظار داشته باشید.

اگر مصرف‌کنندهٔ جدید به سیستم پیام‌رسانی اضافه کنید، معمولاً فقط پیام‌های ارسال‌شده پس از زمان ثبت‌نام را می‌گیرد؛ پیام‌های قبلی رفته‌اند و قابل بازیابی نیستند. این را با فایل‌ها و پایگاه‌های داده مقایسه کنید: می‌توانید هر زمان کلاینت جدید اضافه کنید و داده‌ای را بخوانید که خیلی قبل نوشته شده (تا وقتی برنامه صریحاً overwrite یا حذف نکرده باشد).

چرا ترکیبی نداشته باشیم که رویکرد ذخیرهٔ پایدار پایگاه‌های داده را با امکانات اعلان کم‌تأخیر پیام‌رسانی ترکیب کند؟ این ایدهٔ message broker مبتنی بر لاگ است.

استفاده از لاگ برای ذخیرهٔ پیام

لاگ ساده توالی append-only رکوردها روی دیسک است. قبلاً در زمینهٔ موتورهای ذخیره‌سازی log-structured و write-ahead log در فصل ۳، و در زمینهٔ replication در فصل ۵ دربارهٔ لاگ بحث کردیم.

همان ساختار می‌تواند message broker را پیاده کند: تولیدکننده با append به انتهای لاگ پیام می‌فرستد و مصرف‌کننده با خواندن متوالی لاگ پیام می‌گیرد. اگر مصرف‌کننده به انتهای لاگ برسد، منتظر اعلان append پیام جدید می‌ماند. ابزار یونیکس tail -f که فایل را برای دادهٔ append‌شده رصد می‌کند اساساً همین‌طور کار می‌کند.

برای مقیاس به throughput بالاتر از یک دیسک، لاگ را می‌توان پارتیشن‌بندی کرد (به معنای فصل ۶). پارتیشن‌های مختلف روی ماشین‌های مختلف می‌توانند host شوند، طوری که هر پارتیشن لاگ جداگانه‌ای است که مستقل از بقیه خوانده و نوشته می‌شود. topic را می‌توان به‌عنوان گروهی از پارتیشن‌ها تعریف کرد که همه پیام‌های یک نوع را حمل می‌کنند. این رویکرد در شکل ۱۱-۳ نشان داده شده.

در هر پارتیشن، broker به هر پیام شمارهٔ توالی monotonic افزایشی، یا offset، می‌دهد (در شکل ۱۱-۳، اعداد در جعبه‌ها offset پیام‌اند). چنین شمارهٔ توالی منطقی است چون پارتیشن append-only است و پیام‌ها در پارتیشن کاملاً مرتب‌اند. تضمین ترتیب بین پارتیشن‌های مختلف وجود ندارد.

شکل ۱۱-۳. تولیدکنندگان با append به فایل topic-partition پیام می‌فرستند و مصرف‌کنندگان این فایل‌ها را متوالی می‌خوانند.

Apache Kafka [۱۷، ۱۸]، Amazon Kinesis Streams [۱۹] و DistributedLog توییتر [۲۰، ۲۱] message brokerهای مبتنی بر لاگ هستند که این‌طور کار می‌کنند. Google Cloud Pub/Sub از نظر معماری مشابه است اما API سبک JMS را به‌جای abstraction لاگ expose می‌کند [۱۶]. اگرچه این brokerها همهٔ پیام‌ها را روی دیسک می‌نویسند، با پارتیشن‌بندی روی چند ماشین و replication پیام‌ها به throughput میلیون‌ها پیام در ثانیه می‌رسند [۲۲، ۲۳].

لاگ‌ها در مقایسه با پیام‌رسانی سنتی

رویکرد مبتنی بر لاگ به‌سادگی fan-out را پشتیبانی می‌کند چون چند مصرف‌کننده می‌توانند مستقل لاگ را بخوانند بدون تأثیر بر یکدیگر — خواندن پیام آن را از لاگ حذف نمی‌کند. برای load balancing در گروه مصرف‌کنندگان، به‌جای تخصیص پیام‌های تکی به کلاینت‌ها، broker می‌تواند کل پارتیشن‌ها را به گره‌های گروه مصرف‌کننده تخصیص دهد.

هر کلاینت سپس همهٔ پیام‌های پارتیشن‌های تخصیص‌یافته را مصرف می‌کند. معمولاً وقتی مصرف‌کننده پارتیشن لاگ گرفته، پیام‌های پارتیشن را متوالی و تک‌نخی می‌خواند. این load balancing درشت‌دانه معایبی دارد:

• تعداد گره‌هایی که کار مصرف یک topic را تقسیم می‌کنند حداکثر به تعداد پارتیشن‌های لاگ آن topic محدود است، چون پیام‌های یک پارتیشن به همان گره تحویل می‌شوند.i
• اگر یک پیام کند پردازش شود، پردازش پیام‌های بعدی همان پارتیشن را متوقف می‌کند (نوعی head-of-line blocking؛ به «توصیف کارایی» در صفحه ۱۳ مراجعه کنید).

بنابراین در مواردی که پیام‌ها گران پردازش‌اند و می‌خواهید پردازش را پیام‌به‌پیام موازی کنید و ترتیب پیام مهم نیست، سبک JMS/AMQP message broker ترجیح دارد. در مواردی با throughput بالا، پردازش سریع هر پیام و اهمیت ترتیب پیام، رویکرد مبتنی بر لاگ بسیار خوب کار می‌کند.

i. می‌توان طرح load balancing ساخت که دو مصرف‌کننده کار پردازش یک پارتیشن را با خواندن کل مجموعهٔ پیام‌ها تقسیم کنند، اما یکی فقط offsetهای زوج و دیگری فرد را پردازش کند. یا پردازش را روی thread pool پخش کرد، اما مدیریت offset مصرف‌کننده پیچیده‌تر می‌شود. به‌طور کلی پردازش تک‌نخی پارتیشن ترجیح دارد و موازی‌سازی با پارتیشن‌های بیشتر افزایش می‌یابد.

offset مصرف‌کننده

مصرف متوالی پارتیشن تشخیص پیام‌های پردازش‌شده را آسان می‌کند: همهٔ پیام‌هایی با offset کمتر از offset فعلی مصرف‌کننده پردازش شده‌اند و همهٔ پیام‌های با offset بزرگ‌تر هنوز دیده نشده‌اند. بنابراین broker نیازی به ردیابی acknowledgment هر پیام ندارد — فقط دوره‌ای offset مصرف‌کننده را ثبت می‌کند. سربار کمتر bookkeeping و فرصت‌های batching و pipelining در این رویکرد throughput سیستم‌های مبتنی بر لاگ را بالا می‌برد.

این offset بسیار شبیه log sequence number رایج در replication تک-leader پایگاه داده است که در «راه‌اندازی followerهای جدید» در صفحه ۱۵۵ بحث شد. در replication پایگاه داده، log sequence number به follower اجازه می‌دهد پس از قطع به leader وصل شود و replication را بدون رد شدن write از سر بگیرد. همان اصل اینجا به کار می‌رود: message broker مانند leader پایگاه داده رفتار می‌کند و مصرف‌کننده مانند follower.

اگر گره مصرف‌کننده خراب شود، گره دیگری در گروه مصرف‌کننده پارتیشن‌های مصرف‌کنندهٔ خراب را می‌گیرد و از آخرین offset ثبت‌شده مصرف را شروع می‌کند. اگر مصرف‌کننده پیام‌های بعدی را پردازش کرده اما offset آن‌ها را ثبت نکرده باشد، آن پیام‌ها پس از restart دوباره پردازش می‌شوند. بعد در فصل راه‌های برخورد با این مسئله را بحث می‌کنیم.

مصرف فضای دیسک

اگر فقط به لاگ append کنید، سرانجام فضای دیسک تمام می‌شود. برای آزاد کردن فضا، لاگ به segment تقسیم می‌شود و گاه segmentهای قدیمی حذف یا به ذخیرهٔ آرشیو منتقل می‌شوند. (بعد روش پیشرفته‌تر آزاد کردن فضای دیسک را بحث می‌کنیم.)

یعنی اگر مصرف‌کنندهٔ کند از نرخ پیام عقب بماند و آنقدر عقب بیفتد که offsetش به segment حذف‌شده اشاره کند، برخی پیام‌ها را از دست می‌دهد. در عمل، لاگ buffer با اندازهٔ محدود پیاده می‌کند که پیام‌های قدیمی را هنگام پر شدن دور می‌ریزد — circular buffer یا ring buffer هم نامیده می‌شود. با این حال چون buffer روی دیسک است، می‌تواند بسیار بزرگ باشد.

یک محاسبهٔ تقریبی: در زمان نگارش، دیسک سخت بزرگ معمولی ظرفیت ۶ TB و throughput نوشتن متوالی ۱۵۰ MB/s دارد. اگر با بیشترین نرخ ممکن پیام بنویسید، پر کردن دیسک حدود ۱۱ ساعت طول می‌کشد. پس دیسک می‌تواند ۱۱ ساعت پیام buffer کند و سپس پیام‌های قدیمی را overwrite کند. این نسبت حتی با دیسک و ماشین زیاد ثابت می‌ماند. در عمل، استقرارها به‌ندرت از کل پهنای باند نوشتن دیسک استفاده می‌کنند، پس لاگ معمولاً چند روز یا حتی هفته پیام نگه می‌دارد.

صرف‌نظر از مدت نگه‌داری پیام، throughput لاگ تقریباً ثابت می‌ماند چون هر پیام به هر حال روی دیسک نوشته می‌شود [۱۸]. این با سیستم‌های پیام‌رسانی که پیش‌فرض پیام را در حافظه نگه می‌دارند و فقط اگر صف خیلی بزرگ شد روی دیسک می‌نویسند متفاوت است: چنین سیستم‌هایی وقتی صف کوتاه است سریع‌اند و با نوشتن روی دیسک خیلی کندتر می‌شوند، پس throughput به مقدار تاریخ نگه‌داری‌شده بستگی دارد.

وقتی مصرف‌کنندگان از تولیدکنندگان عقب می‌مانند

در ابتدای «سیستم‌های پیام‌رسانی» در صفحه ۴۴۱ سه گزینه را برای وقتی که مصرف‌کننده از نرخ تولیدکنندگان عقب می‌ماند بحث کردیم: drop پیام، buffering یا backpressure. در این طبقه‌بندی، رویکرد مبتنی بر لاگ شکلی از buffering با buffer بزرگ اما با اندازهٔ ثابت است (محدود به فضای دیسک موجود).

اگر مصرف‌کننده آنقدر عقب بیفتد که پیام‌های موردنیازش قدیمی‌تر از آنچه روی دیسک نگه داشته شده باشند، نمی‌تواند آن‌ها را بخواند — پس broker عملاً پیام‌های قدیمی‌تر از ظرفیت buffer را drop می‌کند. می‌توانید فاصلهٔ مصرف‌کننده از head لاگ را مانیتور کنید و اگر به‌طور قابل‌توجهی عقب افتاد هشدار دهید. چون buffer بزرگ است، زمان کافی برای اپراتور انسانی برای رفع مصرف‌کنندهٔ کند و جبران عقب‌ماندگی قبل از از دست رفتن پیام وجود دارد.

حتی اگر مصرف‌کننده خیلی عقب بیفتد و پیام از دست بدهد، فقط همان مصرف‌کننده تحت تأثیر است؛ سرویس سایر مصرف‌کنندگان مختل نمی‌شود. این مزیت عملیاتی بزرگی است: می‌توانید برای توسعه، تست یا دیباگ آزمایشی از لاگ production مصرف کنید بدون نگرانی زیاد از اختلال سرویس production. وقتی مصرف‌کننده خاموش یا crash می‌شود، مصرف منابع متوقف می‌شود — تنها چیزی که می‌ماند offset مصرف‌کننده است.

این رفتار با message brokerهای سنتی هم متفاوت است: باید مراقب باشید صف‌هایی که مصرف‌کننده‌شان خاموش شده حذف شوند — وگرنه بی‌جهت پیام جمع می‌کنند و حافظه را از مصرف‌کنندگان فعال می‌گیرند.

پخش مجدد پیام‌های قدیمی

قبلاً گفتیم در message brokerهای سبک AMQP و JMS، پردازش و تأیید پیام عملی مخرب است چون پیام‌ها را از broker حذف می‌کند. در message broker مبتنی بر لاگ، مصرف پیام بیشتر شبیه خواندن از فایل است: عملی فقط‌خواندنی است که لاگ را تغییر نمی‌دهد.

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

این جنبه message broker مبتنی بر لاگ را بیشتر شبیه فرایندهای دسته‌ای فصل قبل می‌کند، جایی که دادهٔ مشتق‌شده از ورودی با فرایند تبدیل قابل تکرار جداست. آزمایش بیشتر و بازیابی آسان‌تر از خطا و باگ را ممکن می‌کند و ابزار خوبی برای یکپارچه‌سازی جریان‌های داده در سازمان است [۲۴].

پایگاه‌های داده و جریان‌ها

مقایسه‌هایی بین message brokerها و پایگاه‌های داده کشیدیم. اگرچه سنتی ابزارهای جدا دیده می‌شدند، دیدیم message brokerهای مبتنی بر لاگ با گرفتن ایده از پایگاه‌های داده در پیام‌رسانی موفق بوده‌اند. می‌توانیم برعکس هم عمل کنیم: ایده از پیام‌رسانی و جریان‌ها را به پایگاه‌های داده ببریم.

قبلاً گفتیم رویداد رکوردی از چیزی است که در زمانی رخ داده. آنچه رخ داده ممکن است عمل کاربر (مثلاً تایپ query جستجو)، خوانش سنسور یا write به پایگاه داده باشد. اینکه چیزی به پایگاه داده نوشته شده خود رویدادی است که می‌توان capture، ذخیره و پردازش کرد. این مشاهده نشان می‌دهد ارتباط پایگاه‌های داده و جریان‌ها عمیق‌تر از ذخیرهٔ فیزیکی لاگ روی دیسک است — اساساً بنیادی است.

در واقع، replication log (به «پیاده‌سازی لاگ‌های replication» در صفحه ۱۵۸ مراجعه کنید) جریانی از رویدادهای write پایگاه داده است که leader هنگام پردازش تراکنش‌ها تولید می‌کند. followerها آن جریان write را روی کپی خود اعمال می‌کنند و به کپی دقیق همان داده می‌رسند. رویدادهای replication log تغییرات داده را توصیف می‌کنند.

همچنین در «پخش ترتیب کلی» در صفحه ۳۴۸ به اصل state machine replication برخوردیم: اگر هر رویداد write به پایگاه داده را نشان دهد و هر replica همان رویدادها را به همان ترتیب پردازش کند، replicaها به همان وضعیت نهایی می‌رسند. (فرض می‌شود پردازش رویداد قطعی است.) فقط مورد دیگری از جریان‌های رویداد است!

در این بخش ابتدا مسئله‌ای در سیستم‌های داده ناهمگن را می‌بینیم و سپس حل آن با آوردن ایده از جریان‌های رویداد به پایگاه‌های داده را بررسی می‌کنیم.

همگام نگه‌داشتن سیستم‌ها

در سراسر این کتاب دیدیم هیچ سیستم واحدی همهٔ نیازهای ذخیره، query و پردازش داده را برآورده نمی‌کند. در عمل، بیشتر برنامه‌های غیرساده برای نیازهایشان چند فناوری مختلف ترکیب می‌کنند: مثلاً پایگاه OLTP برای درخواست کاربر، cache برای سرعت درخواست‌های رایج، ایندکس full-text برای جستجو و انبار داده برای analytics. هر کدام کپی خود از داده را در نمایشی بهینه‌شده برای هدف خود دارد.

وقتی همان داده یا دادهٔ مرتبط در چند جا ظاهر می‌شود، باید با یکدیگر همگام بمانند: اگر آیتمی در پایگاه داده به‌روز شود، باید در cache، ایندکس جستجو و انبار داده هم به‌روز شود. برای انبار داده این همگام‌سازی معمولاً با فرایندهای ETL انجام می‌شود (به «انبارداری داده» در صفحه ۹۱ مراجعه کنید)، اغلب با گرفتن کپی کامل پایگاه داده، تبدیل و bulk-load به انبار داده — یعنی فرایند دسته‌ای. مشابه، در «خروجی گردش‌کارهای دسته‌ای» در صفحه ۴۱۱ دیدیم ایندکس جستجو، سیستم توصیه و سایر سیستم‌های مشتق‌شده چگونه با فرایند دسته‌ای ساخته می‌شوند.

اگر dump دوره‌ای کامل پایگاه داده خیلی کند باشد، گاه از dual write استفاده می‌شود: کد برنامه صریحاً هنگام تغییر داده به هر سیستم می‌نویسد — مثلاً اول پایگاه داده، بعد ایندکس جستجو، بعد invalidate کردن cache (یا حتی همزمان).

با این حال dual write مشکلات جدی دارد، یکی race condition نشان‌داده‌شده در شکل ۱۱-۴. در این مثال، دو کلاینت همزمان می‌خواهند آیتم X را به‌روز کنند: کلاینت ۱ مقدار A و کلاینت ۲ مقدار B. هر دو اول به پایگاه داده و بعد به ایندکس جستجو می‌نویسند. به‌خاطر زمان‌بندی بد، درخواست‌ها در هم می‌آیند: پایگاه داده اول write کلاینت ۱ (A) و بعد کلاینت ۲ (B) را می‌بیند، پس مقدار نهایی B است. ایندکس جستجو اول write کلاینت ۲ و بعد کلاینت ۱ را می‌بیند، پس مقدار نهایی A است. دو سیستم برای همیشه ناهمگام‌اند، حتی بدون خطا.

شکل ۱۱-۴. در پایگاه داده، X اول A و بعد B می‌شود، در حالی که در ایندکس جستجو writeها به ترتیب معکوس می‌رسند.

مگر مکانیزم تشخیص همزمانی اضافی داشته باشید، مثل version vectorهای «تشخیص writeهای همزمان» در صفحه ۱۸۴، حتی متوجه write همزمان نمی‌شوید — یک مقدار بی‌صدا مقدار دیگر را overwrite می‌کند.

مشکل دیگر dual write این است که یکی از writeها ممکن است شکست بخورد و دیگری موفق شود. این مسئلهٔ تحمل خطا است نه همزمانی، اما باز هم دو سیستم را ناهمگام می‌کند. اطمینان از موفق یا ناموفق بودن هر دو، مسئلهٔ atomic commit است که حل آن گران است (به «Atomic Commit و Two-Phase Commit (2PC)» در صفحه ۳۵۴ مراجعه کنید).

اگر فقط یک پایگاه داده replicated با leader واحد داشته باشید، آن leader ترتیب writeها را تعیین می‌کند و state machine replication بین replicaهای پایگاه داده کار می‌کند. اما در شکل ۱۱-۴ leader واحد نیست: پایگاه داده leader دارد و ایندکس جستجو leader دارد، اما هیچ‌کدام از دیگری پیروی نمی‌کند و تضاد ممکن است (به «رپلیکاسیون چند-leader» در صفحه ۱۶۸ مراجعه کنید).

وضعیت بهتر می‌شد اگر واقعاً فقط یک leader بود — مثلاً پایگاه داده — و ایندکس جستجو follower پایگاه داده می‌شد. اما آیا در عمل ممکن است؟

Capture دادهٔ تغییر (CDC)

مشکل بیشتر لاگ‌های replication پایگاه‌های داده این است که مدت‌ها جزئیات پیاده‌سازی داخلی پایگاه داده تلقی می‌شدند، نه API عمومی. کلاینت‌ها باید از مدل داده و زبان query پایگاه داده query بزنند، نه لاگ replication را parse کنند و داده استخراج کنند.

دهه‌ها بسیاری از پایگاه‌های داده راه مستند برای گرفتن لاگ تغییرات نداشتند. بنابراین سخت بود همهٔ تغییرات پایگاه داده را replicate کرد و به فناوری ذخیره‌سازی دیگر مثل ایندکس جستجو، cache یا انبار داده منتقل کرد.

اخیراً علاقه به change data capture (CDC) — فرایند مشاهدهٔ همهٔ تغییرات داده نوشته‌شده به پایگاه داده و استخراج آن‌ها به شکلی قابل replicate به سیستم‌های دیگر — رشد کرده. CDC به‌ویژه جالب است اگر تغییرات به‌صورت جریان، بلافاصله پس از نوشتن، در دسترس باشند.

مثلاً می‌توانید تغییرات پایگاه داده را capture کنید و همان تغییرات را پیوسته روی ایندکس جستجو اعمال کنید. اگر لاگ تغییرات به همان ترتیب اعمال شود، انتظار دارید دادهٔ ایندکس جستجو با پایگاه داده مطابقت کند. ایندکس جستجو و سایر سیستم‌های مشتق‌شده فقط مصرف‌کنندهٔ جریان تغییرند، همان‌طور که شکل ۱۱-۵ نشان می‌دهد.

شکل ۱۱-۵. گرفتن داده به ترتیبی که به یک پایگاه داده نوشته شده و اعمال تغییرات به سیستم‌های دیگر به همان ترتیب.

پیاده‌سازی change data capture

می‌توانیم مصرف‌کنندگان لاگ را سیستم‌های دادهٔ مشتق‌شده بنامیم، همان‌طور که در مقدمهٔ بخش سوم بحث شد: دادهٔ ایندکس جستجو و انبار داده فقط نمای دیگری از دادهٔ system of record است. CDC مکانیزمی است برای اطمینان از اینکه همهٔ تغییرات system of record در سیستم‌های مشتق‌شده هم منعکس شوند تا کپی دقیق داشته باشند.

اساساً CDC یک پایگاه داده را leader (که تغییرات از آن capture می‌شود) می‌کند و بقیه را follower. message broker مبتنی بر لاگ برای حمل رویدادهای تغییر از پایگاه مبدأ مناسب است چون ترتیب پیام‌ها را حفظ می‌کند (اجتناب از مسئلهٔ جابه‌جایی ترتیب شکل ۱۱-۲).

database trigger می‌تواند CDC را پیاده کند (به «رپلیکاسیون مبتنی بر trigger» در صفحه ۱۶۱ مراجعه کنید) با ثبت triggerهایی که همهٔ تغییرات جداول داده را می‌بینند و ورودی متناظر به جدول changelog اضافه می‌کنند. اما معمولاً شکننده‌اند و سربار کارایی قابل‌توجه دارند. parse کردن replication log رویکرد مقاوم‌تری است، اگرچه چالش‌هایی مثل مدیریت تغییر schema دارد.

Databus لینکدین [۲۵]، Wormhole فیسبوک [۲۶] و Sherpa یاهو! [۲۷] این ایده را در مقیاس بزرگ به کار می‌برند. Bottled Water CDC برای PostgreSQL با APIای که write-ahead log را decode می‌کند پیاده می‌کند [۲۸]، Maxwell و Debezium برای MySQL با parse کردن binlog [۲۹، ۳۰، ۳۱]، Mongoriver oplog مونگوDB را می‌خواند [۳۲، ۳۳] و GoldenGate امکانات مشابه برای Oracle دارد [۳۴، ۳۵].

مانند message brokerها، CDC معمولاً ناهمگام است: پایگاه system of record منتظر اعمال تغییر روی مصرف‌کنندگان قبل از commit نمی‌ماند. این مزیت عملیاتی دارد که مصرف‌کنندهٔ کند زیاد روی system of record اثر نمی‌گذارد، اما معایب همهٔ مسائل replication lag را دارد (به «مشکلات تأخیر رپلیکاسیون» در صفحه ۱۶۱ مراجعه کنید).

snapshot اولیه

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

ساخت ایندکس full-text جدید مثلاً به کپی کامل کل پایگاه داده نیاز دارد — فقط اعمال لاگ تغییرات اخیر کافی نیست چون آیتم‌هایی که اخیراً به‌روز نشده‌اند غایب‌اند. پس اگر کل تاریخچهٔ لاگ را ندارید، باید با snapshot سازگار شروع کنید، همان‌طور که در «راه‌اندازی followerهای جدید» در صفحه ۱۵۵ بحث شد.

snapshot پایگاه داده باید به موقعیت یا offset شناخته‌شده‌ای در لاگ تغییر مربوط باشد تا بدانید پس از پردازش snapshot از کجا تغییرات را اعمال کنید. برخی ابزارهای CDC این snapshot را یکپارچه دارند، برخی دستی می‌گذارند.

فشرده‌سازی لاگ

اگر فقط تاریخچهٔ محدود لاگ نگه دارید، هر بار افزودن سیستم مشتق‌شدهٔ جدید باید فرایند snapshot را طی کنید. اما log compaction جایگزین خوبی است.

log compaction را قبلاً در «ایندکس‌های hash» در صفحه ۷۲، در زمینهٔ موتورهای log-structured (شکل ۳-۲ را ببینید) بحث کردیم. اصل ساده است: موتور ذخیره‌سازی دوره‌ای رکوردهای لاگ با همان کلید را می‌یابد، تکراری‌ها را دور می‌ریزد و فقط آخرین به‌روزرسانی هر کلید را نگه می‌دارد. این فشرده‌سازی و ادغام در پس‌زمینه اجرا می‌شود.

در موتور log-structured، به‌روزرسانی با مقدار null ویژه (tombstone) حذف کلید را نشان می‌دهد و در فشرده‌سازی لاگ باعث حذف می‌شود. اما تا وقتی کلید overwrite یا حذف نشود، در لاگ برای همیشه می‌ماند. فضای دیسک چنین لاگ فشرده به محتوای فعلی پایگاه داده بستگی دارد، نه تعداد writeهای تاریخی. اگر کلید مکرر overwrite شود، مقادیر قبلی جمع‌آوری زباله می‌شوند و فقط آخرین مقدار می‌ماند.

همان ایده در message broker مبتنی بر لاگ و CDC کار می‌کند. اگر CDC طوری تنظیم شود که هر تغییر primary key داشته باشد و هر به‌روزرسانی برای کلید مقدار قبلی همان کلید را جایگزین کند، نگه‌داری فقط آخرین write برای کل خاص کافی است.

اکنون هر وقت بخواهید سیستم مشتق‌شده مثل ایندکس جستجو را بازسازی کنید، می‌توانید مصرف‌کنندهٔ جدید از offset ۰ topic فشرده‌شدهٔ لاگ شروع کنید و همهٔ پیام‌های لاگ را متوالی اسکن کنید. لاگ تضمین می‌کند آخرین مقدار هر کلید در پایگاه داده را دارد (و شاید مقادیر قدیمی‌تر) — یعنی می‌توانید بدون snapshot دیگر از پایگاه مبدأ CDC، کپی کامل محتوای پایگاه داده را بگیرید.

این ویژگی log compaction در Apache Kafka پشتیبانی می‌شود. همان‌طور که بعد می‌بینیم، به message broker اجازه می‌دهد برای ذخیرهٔ پایدار استفاده شود، نه فقط پیام‌رسانی گذرا.

پشتیبانی API برای جریان‌های تغییر

به‌طور فزاینده پایگاه‌های داده جریان‌های تغییر را به‌عنوان رابط درجهٔ اول پشتیبانی می‌کنند، نه تلاش‌های CDC معمولی retrofit. مثلاً RethinkDB به queryها اجازه می‌دهد هنگام تغییر نتایج query اعلان subscribe کنند [۳۶]، Firebase [۳۷] و CouchDB [۳۸] همگام‌سازی داده بر پایهٔ change feed در دسترس برنامه‌ها را فراهم می‌کنند و Meteor از oplog مونگوDB برای subscribe تغییرات داده و به‌روزرسانی UI استفاده می‌کند [۳۹].

VoltDB به تراکنش‌ها اجازه می‌دهد داده را پیوسته از پایگاه داده به‌صورت جریان export کنند [۴۰]. پایگاه داده جریان خروجی را در مدل رابطه‌ای به‌عنوان جدولی نشان می‌دهد که تراکنش‌ها tuple در آن insert می‌کنند اما query نمی‌شود. جریان سپس لاگ tupleهایی است که تراکنش‌های commit‌شده به این جدول ویژه نوشته‌اند، به ترتیب commit. مصرف‌کنندگان خارجی این لاگ را ناهمگام مصرف و برای به‌روزرسانی سیستم‌های مشتق‌شده به کار می‌برند.

Kafka Connect [۴۱] تلاشی برای یکپارچه‌سازی ابزارهای CDC طیف گسترده‌ای از پایگاه‌های داده با Kafka است. وقتی جریان رویدادهای تغییر در Kafka باشد، می‌توان برای به‌روزرسانی سیستم‌های مشتق‌شده مثل ایندکس جستجو و همچنین تغذیه سیستم‌های پردازش جریانی (بعد در فصل) استفاده کرد.

Event Sourcing

موازی‌هایی بین ایده‌های اینجا و event sourcing وجود دارد — تکنیکی که در جامعهٔ domain-driven design (DDD) توسعه یافته [۴۲، ۴۳، ۴۴]. event sourcing را به‌اختصار بحث می‌کنیم چون ایده‌های مفیدی برای سیستم‌های جریانی دارد.

مشابه CDC، event sourcing همهٔ تغییرات وضعیت برنامه را به‌عنوان لاگ رویدادهای تغییر ذخیره می‌کند. بزرگ‌ترین تفاوت این است که event sourcing ایده را در سطح abstraction متفاوتی اعمال می‌کند:

• در CDC، برنامه از پایگاه داده به‌شکل mutable استفاده می‌کند و رکوردها را به‌روز و حذف می‌کند. لاگ تغییرات در سطح پایین از پایگاه داده استخراج می‌شود (مثلاً با parse کردن replication log) که ترتیب writeهای استخراج‌شده با ترتیب واقعی نوشتن مطابقت کند و از race condition شکل ۱۱-۴ اجتناب کند. برنامه‌ای که به پایگاه داده می‌نویسد نیازی به آگاهی از CDC ندارد.
• در event sourcing، منطق برنامه صریحاً بر پایهٔ رویدادهای تغییرناپذیر نوشته‌شده به event log ساخته می‌شود. در این حالت event store فقط append است و به‌روزرسانی یا حذف discouraged یا ممنوع است. رویدادها طوری طراحی می‌شوند که چیزهایی در سطح برنامه را منعکس کنند، نه تغییرات وضعیت سطح پایین.

event sourcing تکنیک قدرتمند مدل‌سازی داده است: از دید برنامه معنادارتر است عمل کاربر را به‌عنوان رویداد تغییرناپذیر ثبت کنیم تا اثر آن عمل روی پایگاه دادهٔ mutable. event sourcing تکامل برنامه را آسان‌تر، دیباگ را با فهمیدن بعداً چرا چیزی رخ داد ساده‌تر و در برابر باگ برنامه محافظت می‌کند (به «مزایای رویدادهای تغییرناپذیر» در صفحه ۴۶۰ مراجعه کنید).

مثلاً ذخیرهٔ رویداد «دانشجو ثبت‌نام دوره را لغو کرد» به‌وضوح قصد یک عمل را بی‌طرف بیان می‌کند، در حالی که اثر جانبی «یک ورودی از جدول enrollments حذف شد و یک دلیل لغو به جدول بازخورد دانشجو اضافه شد» فرض‌های زیادی دربارهٔ استفادهٔ بعدی داده دارد. اگر ویژگی جدید معرفی شود — مثلاً «جای به نفر بعدی لیست انتظار داده شود» — رویکرد event sourcing اجازه می‌دهد اثر جانبی جدید به‌راحتی به رویداد موجود زنجیر شود.

event sourcing شبیه مدل دادهٔ chronicle است [۴۵] و شباهت‌هایی بین event log و fact table در star schema وجود دارد (به «ستاره و برفک: schemaهای analytics» در صفحه ۹۳ مراجعه کنید).

پایگاه‌های دادهٔ تخصصی مثل Event Store [۴۶] برای برنامه‌های event sourcing توسعه یافته‌اند، اما رویکرد به ابزار خاص وابسته نیست. پایگاه دادهٔ معمولی یا message broker مبتنی بر لاگ هم می‌تواند برای ساخت برنامه به این سبک به کار رود.

استخراج وضعیت فعلی از event log

event log به‌تنهایی خیلی مفید نیست چون کاربران معمولاً وضعیت فعلی سیستم را می‌خواهند، نه تاریخچهٔ تغییرات. مثلاً در وب‌سایت فروش، کاربر انتظار محتوای فعلی سبد خرید را دارد، نه لیست append-only همهٔ تغییراتی که به سبد داده.

پس برنامه‌های event sourcing باید لاگ رویدادها (دادهٔ نوشته‌شده به سیستم) را به وضعیت برنامهٔ مناسب نمایش به کاربر تبدیل کنند (نحوهٔ خواندن داده از سیستم [۴۷]). این تبدیل می‌تواند منطق دلخواه داشته باشد، اما باید قطعی باشد تا دوباره اجرا شود و همان وضعیت از event log استخراج شود.

مانند CDC، پخش مجدد event log وضعیت فعلی سیستم را بازسازی می‌کند. اما log compaction متفاوت مدیریت می‌شود:

• رویداد CDC برای به‌روزرسانی رکورد معمولاً کل نسخهٔ جدید رکورد را دارد، پس مقدار فعلی برای primary key کاملاً با آخرین رویداد آن کلید تعیین می‌شود و log compaction می‌تواند رویدادهای قبلی همان کلید را دور بریزد.
• در event sourcing، رویدادها در سطح بالاتر مدل‌سازی می‌شوند: رویداد معمولاً قصد عمل کاربر را بیان می‌کند، نه مکانیک به‌روزرسانی وضعیت ناشی از عمل. در این حالت رویدادهای بعدی معمولاً رویدادهای قبلی را override نمی‌کنند و برای بازسازی وضعیت نهایی به تاریخچهٔ کامل رویدادها نیاز است. log compaction به همان شکل ممکن نیست.

برنامه‌های event sourcing معمولاً مکانیزمی برای ذخیرهٔ snapshot وضعیت فعلی مشتق‌شده از لاگ رویداد دارند تا مجبور نباشند هر بار کل لاگ را دوباره پردازش کنند. اما این فقط بهینه‌سازی کارایی برای سرعت خواندن و بازیابی از crash است؛ قصد این است که سیستم بتواند همهٔ رویدادهای خام را برای همیشه نگه دارد و هر وقت لازم بود کل event log را دوباره پردازش کند. این فرض را در «محدودیت‌های تغییرناپذیری» در صفحه ۴۶۳ بحث می‌کنیم.

دستورات و رویدادها

فلسفهٔ event sourcing با دقت بین رویداد و دستور (command) تمایز قائل می‌شود [۴۸]. وقتی درخواست کاربر می‌رسد، ابتدا دستور است: ممکن است هنوز شکست بخورد، مثلاً به‌خاطر نقض شرط یکپارچگی. برنامه باید اول اعتبار اجرای دستور را بسنجد. اگر اعتبارسنجی موفق و دستور پذیرفته شد، رویداد می‌شود — پایدار و تغییرناپذیر.

مثلاً اگر کاربر بخواهد نام کاربری خاصی ثبت کند یا صندلی هواپیما یا تئاتر رزرو کند، برنامه باید ببیند نام کاربری یا صندلی قبلاً گرفته نشده (این مثال را در «اجماع تحمل‌پذیر خطا» در صفحه ۳۶۴ دیدیم). پس از موفقیت بررسی، برنامه رویدادی تولید می‌کند که نام کاربری توسط user ID مشخص ثبت شد یا صندلی برای مشتری مشخص رزرو شد.

در لحظهٔ تولید رویداد، واقعیت می‌شود. حتی اگر مشتری بعداً رزرو را تغییر یا لغو کند، واقعیت می‌ماند که قبلاً رزرو داشت و تغییر یا لغو رویداد جداگانه‌ای است که بعد اضافه می‌شود.

مصرف‌کنندهٔ جریان رویداد اجازه ندارد رویداد را رد کند: وقتی رویداد را می‌بیند، بخش تغییرناپذیر لاگ شده و شاید مصرف‌کنندگان دیگر هم دیده باشند. پس اعتبارسنجی دستور باید همزمان، قبل از رویداد شدن انجام شود — مثلاً با تراکنش serializable که دستور را اعتبارسنجی و رویداد را publish کند.

یا درخواست رزرو صندلی را به دو رویداد تقسیم کنید: اول رزرو موقت، بعد رویداد تأیید جداگانه پس از اعتبارسنجی (همان‌طور که در «پیاده‌سازی ذخیره‌سازی linearizable با پخش ترتیب کلی» در صفحه ۳۵۰ بحث شد). این تقسیم اعتبارسنجی را در فرایند ناهمگام ممکن می‌کند.

وضعیت، جریان‌ها و تغییرناپذیری

در فصل ۱۰ دیدیم پردازش دسته‌ای از تغییرناپذیری فایل‌های ورودی سود می‌برد؛ می‌توانید jobهای آزمایشی روی فایل‌های ورودی موجود بدون ترس آسیب اجرا کنید. این اصل تغییرناپذیری event sourcing و CDC را هم قدرتمند می‌کند.

معمولاً پایگاه داده را ذخیرهٔ وضعیت فعلی برنامه می‌دانیم — این نمایش برای خواندن بهینه است و معمولاً برای سرویس query راحت‌ترین است. ماهیت وضعیت تغییر است، پس پایگاه‌های داده به‌روزرسانی و حذف و هم insert را پشتیبانی می‌کنند. این چگونه با تغییرناپذیری جور می‌شود؟

هر وقت وضعیتی تغییر کند، نتیجهٔ رویدادهایی است که آن را در طول زمان mutate کرده‌اند. مثلاً لیست صندلی‌های آزاد نتیجهٔ رزروهای پردازش‌شده است، موجودی حساب جمع اعتبار و بدهی‌هاست و نمودار زمان پاسخ وب‌سرور تجمیع زمان پاسخ همهٔ درخواست‌های وب است.

هرچه وضعیت چگونه تغییر کند، همیشه توالی رویدادی بوده که آن تغییرات را ایجاد کرده. حتی وقتی کارها انجام و لغو می‌شوند، واقعیت وقوع آن رویدادها باقی می‌ماند. ایدهٔ کلیدی این است که وضعیت mutable و لاگ append-only رویدادهای تغییرناپذیر متناقض نیستند — دو روی یک سکه‌اند. لاگ همهٔ تغییرات، changelog، تکامل وضعیت در زمان را نشان می‌دهد.

اگر ذهنیت ریاضی دارید، می‌توانید بگویید وضعیت برنامه حاصل انتگرال‌گیری جریان رویداد در زمان است و جریان تغییر حاصل مشتق وضعیت نسبت به زمان، همان‌طور که شکل ۱۱-۶ نشان می‌دهد [۴۹، ۵۰، ۵۱]. تشبیه محدودیت دارد (مثلاً مشتق دوم وضعیت معنادار به نظر نمی‌رسد)، اما نقطهٔ شروع مفیدی برای فکر کردن به داده است.

شکل ۱۱-۶. رابطهٔ وضعیت فعلی برنامه و جریان رویداد.

اگر changelog را پایدار ذخیره کنید، اثرش قابل بازتولید کردن وضعیت است. اگر لاگ رویدادها را system of record بدانید و هر وضعیت mutable مشتق‌شده از آن، استدلال دربارهٔ جریان داده در سیستم آسان‌تر می‌شود. همان‌طور که Pat Helland می‌گوید [۵۲]:

Transaction log همهٔ تغییرات پایگاه داده را ثبت می‌کند. append پرسرعت تنها راه تغییر لاگ است. از این منظر، محتوای پایگاه داده cache آخرین مقادیر رکورد در لاگ‌هاست. حقیقت لاگ است. پایگاه داده cache زیرمجموعه‌ای از لاگ است. آن زیرمجموعهٔ cache‌شده اتفاقاً آخرین مقدار هر رکورد و مقدار ایندکس از لاگ است.

log compaction، همان‌طور که در «فشرده‌سازی لاگ» در صفحه ۴۵۶ بحث شد، راهی برای پل زدن تفاوت لاگ و وضعیت پایگاه داده است: فقط آخرین نسخهٔ هر رکورد را نگه می‌دارد و نسخه‌های overwrite‌شده را دور می‌ریزد.

مزایای رویدادهای تغییرناپذیر

تغییرناپذیری در پایگاه‌های داده ایدهٔ قدیمی است. مثلاً حسابداران قرن‌ها در دفترداری مالی از تغییرناپذیری استفاده کرده‌اند. وقتی تراکنش رخ می‌دهد، در دفتر append-only ثبت می‌شود — اساساً لاگ رویدادهایی که پول، کالا یا خدمات جابه‌جا شده را توصیف می‌کند. حساب‌ها مثل سود و زیان یا ترازنامه از جمع تراکنش‌های دفتر مشتق می‌شوند [۵۳].

اگر اشتباه شود، حسابدار تراکنش نادرست را در دفتر پاک یا تغییر نمی‌دهد — تراکنش جبرانی اضافه می‌کند، مثلاً بازپرداخت شارژ نادرست. تراکنش نادرست برای همیشه در دفتر می‌ماند چون برای audit مهم است. اگر ارقام نادرست مشتق‌شده از دفتر نادرست منتشر شده، دورهٔ حسابداری بعد اصلاح را شامل می‌شود. این فرایند در حسابداری کاملاً عادی است [۵۴].

auditability در سیستم‌های مالی به‌ویژه مهم است، اما برای بسیاری سیستم‌های غیرتحت چنین مقررات سختگیرانه‌ای هم مفید است. همان‌طور که در «فلسفهٔ خروجی فرایند دسته‌ای» در صفحه ۴۱۳ بحث شد، اگر کد باگ‌دار که دادهٔ بد می‌نویسد deploy کنید، بازیابی سخت‌تر است اگر کد بتواند مخرب داده overwrite کند. با لاگ append-only رویدادهای تغییرناپذیر، تشخیص و بازیابی آسان‌تر است.

رویدادهای تغییرناپذیر اطلاعات بیشتری از فقط وضعیت فعلی دارند. مثلاً در وب‌فروش، مشتری ممکن است آیتمی به سبد اضافه و بعد حذف کند. از دید تحویل سفارش رویداد دوم اولی را لغو می‌کند، اما برای analytics مفید است بدانید مشتری آیتم را در نظر گرفته و رد کرده — شاید بعد بخرد یا جایگزین پیدا کند. این در event log ثبت می‌شود اما در پایگاه داده‌ای که با حذف آیتم از سبد پاک می‌کند از بین می‌رود [۴۲].

استخراج چند نما از یک event log

با جدا کردن وضعیت mutable از event log تغییرناپذیر، می‌توانید چند نمای خواندنی مختلف از همان لاگ رویداد مشتق کنید. این مانند چند مصرف‌کننده از یک جریان است (شکل ۱۱-۵): مثلاً پایگاه analytics Druid مستقیماً از Kafka با این رویکرد ingest می‌کند [۵۵]، Pistachio key-value store توزیع‌شده‌ای است که Kafka را به‌عنوان commit log به کار می‌برد [۵۶] و sinkهای Kafka Connect داده را از Kafka به پایگاه‌ها و ایندکس‌های مختلف export می‌کنند [۴۱]. برای بسیاری سیستم‌های ذخیره و ایندکس دیگر مثل search server هم منطقی است ورودی را از لاگ توزیع‌شده بگیرند (به «همگام نگه‌داشتن سیستم‌ها» در صفحه ۴۵۲ مراجعه کنید).

گام ترجمهٔ صریح از event log به پایگاه داده تکامل برنامه را آسان‌تر می‌کند: اگر ویژگی جدیدی بخواهید که دادهٔ موجود را به شکل جدید نشان دهد، می‌توانید از event log نمای خواندنی جدا برای ویژگی جدید بسازید و کنار سیستم‌های موجود اجرا کنید بدون تغییر آن‌ها. اجرای موازی سیستم قدیم و جدید اغلب از migration schema پیچیده در سیستم موجود آسان‌تر است. وقتی سیستم قدیم لازم نبود، خاموش و منابعش آزاد می‌شود [۴۷، ۵۷].

ذخیرهٔ داده معمولاً ساده است اگر نگران نحوهٔ query و دسترسی نباشید؛ بسیاری پیچیدگی‌های طراحی schema، ایندکس و موتور ذخیره‌سازی از پشتیبانی الگوهای query و دسترسی خاص ناشی می‌شود (فصل ۳ را ببینید). پس با جدا کردن شکل نوشتن از شکل خواندن و اجازهٔ چند نمای خواندنی انعطاف زیادی می‌گیرید. این ایده گاه command query responsibility segregation (CQRS) نامیده می‌شود [۴۲، ۵۸، ۵۹].

رویکرد سنتی طراحی پایگاه داده و schema بر این فرض اشتباه است که داده باید به همان شکلی که query می‌شود نوشته شود. بحث نرمال‌سازی و denormalization (به «رابطه‌های چندبه‌یک و چندبه‌چند» در صفحه ۳۳ مراجعه کنید) اگر بتوانید داده را از event log بهینهٔ نوشتن به وضعیت بهینهٔ خواندن ترجمه کنید، تا حد زیادی بی‌ربط می‌شود: denormalize کردن در نماهای بهینهٔ خواندن کاملاً منطقی است چون فرایند ترجمه آن را با event log سازگار نگه می‌دارد.

در «توصیف بار» در صفحه ۱۱ timeline خانهٔ توییتر را دیدیم — cache توییت‌های اخیر افرادی که کاربر دنبال می‌کند (مثل صندوق پستی). این نمونهٔ دیگری از وضعیت بهینهٔ خواندن است: timelineهای خانه بسیار denormalized‌اند چون توییت‌های شما در timeline همهٔ دنبال‌کنندگان تکرار می‌شود. اما fan-out service این وضعیت تکراری را با توییت‌ها و روابط دنبال‌کردن جدید همگام نگه می‌دارد و تکرار را قابل مدیریت می‌کند.

کنترل همزمانی

بزرگ‌ترین معایب event sourcing و CDC این است که مصرف‌کنندگان event log معمولاً ناهمگام‌اند؛ ممکن است کاربر به لاگ بنویسد، از نمای مشتق‌شده بخواند و ببیند write هنوز منعکس نشده. این مسئله و راه‌حل‌های احتمالی را در «خواندن writeهای خودتان» در صفحه ۱۶۲ بحث کردیم.

یک راه‌حل، به‌روزرسانی همزمان نمای خواندن همراه append رویداد به لاگ است. این به تراکنشی نیاز دارد که writeها را در واحد اتمی ترکیب کند، پس یا event log و نمای خواندن در همان سیستم ذخیره‌سازی باشند یا تراکنش توزیع‌شده بین سیستم‌های مختلف. یا از رویکرد «پیاده‌سازی ذخیره‌سازی linearizable با پخش ترتیب کلی» در صفحه ۳۵۰ استفاده کنید.

از سوی دیگر، استخراج وضعیت فعلی از event log برخی جنبه‌های کنترل همزمانی را ساده می‌کند. بخش زیادی از نیاز به تراکنش چندشیءی (به «عملیات تک‌شیءی و چندشیءی» در صفحه ۲۲۸ مراجعه کنید) از این است که یک عمل کاربر نیاز به تغییر داده در چند جا دارد. با event sourcing می‌توانید رویدادی طراحی کنید که عمل کاربر را خودکفا توصیف کند. عمل کاربر فقط یک write در یک جا می‌خواهد — append رویداد به لاگ — که اتمی کردنش آسان است.

اگر event log و وضعیت برنامه به همان شکل پارتیشن‌بندی شوند (مثلاً پردازش رویداد مشتری در پارتیشن ۳ فقط پارتیشن ۳ وضعیت برنامه را به‌روز کند)، مصرف‌کنندهٔ لاگ تک‌نخی ساده برای write به کنترل همزمانی نیاز ندارد — به‌طور سازنده فقط یک رویداد در هر لحظه پردازش می‌کند (به «اجرای واقعاً ترتیبی» در صفحه ۲۵۲ هم مراجعه کنید). لاگ غیرقطعیت همزمانی را با تعریف ترتیب سریال رویدادها در پارتیشن برمی‌دارد [۲۴]. اگر رویداد به چند پارتیشن وضعیت بخورد، کمی کار بیشتر لازم است که در فصل ۱۲ بحث می‌کنیم.

محدودیت‌های تغییرناپذیری

بسیاری سیستم‌هایی که event-sourced نیستند باز هم به تغییرناپذیری تکیه می‌کنند: پایگاه‌های مختلف دروناً ساختار دادهٔ تغییرناپذیر یا دادهٔ چندنسخه‌ای برای snapshot لحظه‌ای دارند (به «ایندکس‌ها و snapshot isolation» در صفحه ۲۴۱ مراجعه کنید). سیستم‌های کنترل نسخه مثل Git، Mercurial و Fossil هم برای حفظ تاریخ نسخهٔ فایل‌ها به دادهٔ تغییرناپذیر وابسته‌اند.

تا چه حد نگه‌داری تاریخ تغییرناپذیر همهٔ تغییرات برای همیشه عملی است؟ به میزان churn در مجموعه‌داده بستگی دارد. برخی بارها عمدتاً داده اضافه می‌کنند و به‌ندرت به‌روز یا حذف می‌کنند — تغییرناپذیر کردنشان آسان است. برخی نرخ بالای به‌روزرسانی و حذف روی مجموعه‌دادهٔ نسبتاً کوچک دارند؛ در این موارد تاریخ تغییرناپذیر ممکن است خیلی بزرگ شود، fragmentation مسئله شود و کارایی compaction و garbage collection برای استحکام عملیاتی حیاتی شود [۶۰، ۶۱].

علاوه بر کارایی، گاه به‌دلایل اداری باید داده حذف شود با وجود تغییرناپذیری. مثلاً مقررات حریم خصوصی ممکن است پس از بستن حساب، اطلاعات شخصی کاربر را بخواهد؛ قوانین حفاظت داده حذف اطلاعات نادرست را می‌خواهد؛ یا نشت تصادفی اطلاعات حساس باید مهار شود. در این موارد کافی نیست فقط رویداد دیگری به لاگ اضافه کنید که دادهٔ قبلی حذف‌شده تلقی شود — باید تاریخ را بازنویسی کنید و وانمود کنید داده هرگز نوشته نشده. مثلاً Datomic این را excision می‌نامد [۶۲] و Fossil مفهوم مشابه shunning دارد [۶۳].

حذف واقعی داده شگفت‌آور سخت است [۶۴] چون کپی‌ها جاهای زیاد می‌مانند: موتورهای ذخیره‌سازی، فایل‌سیستم‌ها و SSDها اغلب به مکان جدید می‌نویسند نه overwrite درجا [۵۲] و backupها عمداً تغییرناپذیرند تا حذف یا خرابی تصادفی نشود. حذف بیشتر «سخت‌تر کردن بازیابی داده» است تا «غیرممکن کردن بازیابی». با این حال گاه باید تلاش کرد، همان‌طور که در «قانون‌گذاری و خودتنظیمی» در صفحه ۵۴۲ می‌بینیم.

پردازش جریان‌ها

تا اینجا دربارهٔ منشأ جریان‌ها (رویدادهای فعالیت کاربر، سنسورها و write به پایگاه‌های داده) و نحوهٔ انتقال (پیام‌رسانی مستقیم، message broker و event log) بحث کردیم. باقی‌مانده این است که با جریان چه کار کنید — یعنی پردازشش. به‌طور کلی سه گزینه دارید:

۱. دادهٔ رویدادها را در پایگاه داده، cache، ایندکس جستجو یا ذخیره‌سازی مشابه بنویسید تا کلاینت‌های دیگر query کنند. همان‌طور که شکل ۱۱-۵ نشان می‌دهد، راه خوبی برای همگام نگه‌داشتن پایگاه داده با تغییرات سایر بخش‌های سیستم است — به‌ویژه اگر مصرف‌کنندهٔ جریان تنها کلاینت نوشتن به پایگاه داده باشد. نوشتن به ذخیره‌سازی معادل جریانی آنچه در «خروجی گردش‌کارهای دسته‌ای» در صفحه ۴۱۱ بحث شد است.

۲. رویدادها را به‌نوعی به کاربران push کنید — مثلاً ایمیل هشدار، push notification یا استریم به داشبورد real-time برای visualization. در این حالت انسان مصرف‌کنندهٔ نهایی جریان است.

۳. یک یا چند جریان ورودی را پردازش کنید تا یک یا چند جریان خروجی تولید شود. جریان‌ها ممکن است از چند مرحلهٔ پردازش عبور کنند تا در نهایت به خروجی (گزینه ۱ یا ۲) برسند.

در بقیهٔ فصل گزینه ۳ را بحث می‌کنیم: پردازش جریان‌ها برای تولید جریان‌های مشتق‌شدهٔ دیگر. قطعه کدی که جریان‌ها را این‌طور پردازش می‌کند operator یا job نامیده می‌شود. به فرایندهای یونیکس و jobهای MapReduce فصل ۱۰ و الگوی جریان داده نزدیک است: پردازندهٔ جریانی جریان‌های ورودی را فقط‌خواندنی مصرف و خروجی را append-only در جای دیگر می‌نویسد.

الگوهای پارتیشن‌بندی و موازی‌سازی در پردازنده‌های جریانی هم بسیار شبیه MapReduce و موتورهای جریان داده فصل ۱۰ است، پس آن موضوعات را تکرار نمی‌کنیم. عملیات نگاشت پایه مثل تبدیل و فیلتر رکوردها هم مشابه کار می‌کند.

تفاوت حیاتی با jobهای دسته‌ای این است که جریان هرگز پایان نمی‌یابد. این تفاوت پیامدهای زیادی دارد: همان‌طور که ابتدای فصل گفتیم، مرتب‌سازی با مجموعه‌دادهٔ نامحدود معنا ندارد و joinهای sort-merge (به «joinها و گروه‌بندی سمت reduce» در صفحه ۴۰۳ مراجعه کنید) قابل استفاده نیستند. مکانیزم‌های تحمل خطا هم باید عوض شوند: در job دسته‌ای چند دقیقه‌ای، restart task شکست‌خورده از ابتدا ساده است، اما در job جریانی چندساله restart از ابتدا پس از crash شاید عملی نباشد.

کاربردهای پردازش جریانی

پردازش جریانی مدت‌ها برای مانیتورینگ به کار رفته — سازمانی می‌خواهد هنگام وقوع برخی چیزها هشدار بگیرد. مثلاً:

• سیستم‌های تشخیص تقلب باید الگوی استفادهٔ کارت اعتباری را ببینند و در صورت سرقت احتمالی کارت را مسدود کنند.
• سیستم‌های معاملاتی تغییر قیمت بازار را بررسی و طبق قواعد معامله اجرا می‌کنند.
• سیستم‌های تولید وضعیت ماشین‌های کارخانه را مانیتور و در خرابی سریع مشکل را شناسایی می‌کنند.
• سیستم‌های نظامی و اطلاعاتی فعالیت مهاجم احتمالی را ردیابی و در نشانهٔ حمله هشدار می‌دهند.

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

پردازش رویداد پیچیده (CEP)

complex event processing (CEP) رویکردی است که در دههٔ ۱۹۹۰ برای تحلیل جریان رویداد، به‌ویژه جستجوی الگوهای رویداد خاص، توسعه یافت [۶۵، ۶۶]. مشابه regex که الگویی از کاراکترها در رشته جستجو می‌کند، CEP قواعدی برای جستجوی الگوهای رویداد در جریان تعریف می‌کند.

سیستم‌های CEP اغلب زبان query سطح بالای declarative مثل SQL یا رابط گرافیکی برای توصیف الگوهای رویداد دارند. queryها به موتور پردازشی داده می‌شود که جریان‌های ورودی را مصرف و دروناً state machine برای تطبیق لازم نگه می‌دارد. با یافتن تطابق، موتور complex event (نام CEP) با جزئیات الگوی یافت‌شده emit می‌کند [۶۷].

در این سیستم‌ها رابطهٔ query و داده برعکس پایگاه‌های داده معمولی است. معمولاً پایگاه داده داده را پایدار نگه می‌دارد و query گذرا است: query می‌آید، پایگاه داده دادهٔ منطبق را جستجو و پس از پایان query را فراموش می‌کند. موتورهای CEP نقش‌ها را معکوس می‌کنند: queryها بلندمدت ذخیره می‌شوند و رویدادهای جریان‌های ورودی پیوسته برای یافتن query منطبق جریان می‌یابند [۶۸].

پیاده‌سازی‌های CEP شامل Esper [۶۹]، IBM InfoSphere Streams [۷۰]، Apama، TIBCO StreamBase و SQLstream است. پردازنده‌های جریانی توزیع‌شده مثل Samza هم پشتیبانی SQL برای queryهای declarative روی جریان‌ها می‌گیرند [۷۱].

تحلیل جریانی

حوزهٔ دیگر استفادهٔ پردازش جریانی برای analytics روی جریان‌هاست. مرز CEP و تحلیل جریانی مبهم است، اما به‌طور کلی analytics کمتر به یافتن توالی رویداد خاص و بیشتر به تجمیع و متریک‌های آماری روی تعداد زیاد رویداد علاقه دارد — مثلاً:

• اندازه‌گیری نرخ نوعی رویداد (چند بار در بازهٔ زمانی)
• میانگین متحرک مقدار در بازهٔ زمانی
• مقایسهٔ آمار فعلی با بازه‌های قبلی (مثلاً تشخیص روند یا هشدار وقتی متریک نسبت به همان زمان هفتهٔ قبل غیرعادی بالا یا پایین است)

چنین آمارها معمولاً روی بازه‌های زمانی ثابت محاسبه می‌شوند — مثلاً میانگین تعداد query در ثانیه به سرویس در ۵ دقیقهٔ گذشته و percentile 99 زمان پاسخ در آن دوره. میانگین‌گیری روی چند دقیقه نوسان‌های بی‌ربط ثانیه‌به‌ثانیه را هموار می‌کند و همزمان تصویر به‌موقع از تغییر الگوی ترافیک می‌دهد. بازهٔ تجمیع window نامیده می‌شود و در «استدلال دربارهٔ زمان» در صفحه ۴۶۸ جزئیات بیشتری می‌بینیم.

سیستم‌های تحلیل جریانی گاه از الگوریتم‌های probabilistic استفاده می‌کنند، مثل Bloom filter (در «بهینه‌سازی‌های کارایی» در صفحه ۷۹ برای عضویت مجموعه)، HyperLogLog [۷۲] برای تخمین cardinality و الگوریتم‌های مختلف تخمین percentile (به «percentile در عمل» در صفحه ۱۶ مراجعه کنید). الگوریتم‌های probabilistic نتایج تقریبی می‌دهند اما حافظهٔ کمتری در پردازندهٔ جریانی می‌خواهند. این گاه باعث باور می‌شود سیستم‌های پردازش جریانی همیشه تقریبی و با اتلاف‌اند، که نادرست است: ذات پردازش جریانی تقریبی نیست و الگوریتم‌های probabilistic فقط بهینه‌سازی‌اند [۷۳].

چارچوب‌های پردازش جریانی توزیع‌شده متن‌باز زیادی برای analytics طراحی شده‌اند: Apache Storm، Spark Streaming، Flink، Concord، Samza و Kafka Streams [۷۴]. سرویس‌های hosted شامل Google Cloud Dataflow و Azure Stream Analytics است.

نگه‌داری نماهای materialized

در «پایگاه‌های داده و جریان‌ها» در صفحه ۴۵۱ دیدیم جریان تغییرات پایگاه داده می‌تواند سیستم‌های مشتق‌شده مثل cache، ایندکس جستجو و انبار داده را با پایگاه مبدأ همگام نگه دارد. این‌ها موارد خاص نگه‌داری materialized view هستند (به «تجمیع: data cube و نماهای materialized» در صفحه ۱۰۱ مراجعه کنید): استخراج نمای جایگزین روی مجموعه‌داده برای query کارآمد و به‌روزرسانی آن با تغییر دادهٔ زیرین [۵۰].

مشابه، در event sourcing وضعیت برنامه با اعمال لاگ رویدادها نگه داشته می‌شود؛ وضعیت برنامه هم نوعی materialized view است. برخلاف سناریوهای analytics جریانی، معمولاً کافی نیست فقط رویدادهای یک بازهٔ زمانی را در نظر بگیرید: ساخت materialized view ممکن است به همهٔ رویدادها در بازهٔ دلخواه زمانی نیاز داشته باشد، به‌جز رویدادهای منسوخ که log compaction دور می‌ریزد (به «فشرده‌سازی لاگ» در صفحه ۴۵۶ مراجعه کنید). در عمل به پنجره‌ای نیاز دارید که تا ابتدای زمان کشیده شود.

در اصل هر پردازندهٔ جریانی می‌تواند برای نگه‌داری materialized view به کار رود، اگرچه نیاز نگه‌داری رویدادها برای همیشه با فرض برخی چارچوب‌های analytics که عمدتاً روی پنجره‌های محدود کار می‌کنند در تضاد است. Samza و Kafka Streams این کاربرد را با تکیه بر log compaction در Kafka پشتیبانی می‌کنند [۷۵].

جستجو روی جریان‌ها

علاوه بر CEP که الگوهای چند رویدادی را جستجو می‌کند، گاه نیاز به جستجوی رویدادهای تکی بر معیارهای پیچیده مثل queryهای full-text است.

مثلاً سرویس‌های پایش رسانه به feed خبر و پخش رسانه‌ها subscribe می‌کنند و خبری که شرکت، محصول یا موضوع مورد علاقه را ذکر کند جستجو می‌کنند. با فرمول‌بندی query جستجو از پیش و تطبیق پیوستهٔ جریان خبر با آن query انجام می‌شود. ویژگی مشابه در برخی وب‌سایت‌ها هست: کاربران املاک می‌توانند هنگام ظاهر شدن ملک جدید مطابق معیارشان مطلع شوند. ویژگی percolator در Elasticsearch [۷۶] یکی از گزینه‌های پیاده‌سازی این جستجوی جریانی است.

موتورهای جستجوی معمولی اول اسناد را ایندکس و بعد روی ایندکس query می‌زنند. در مقابل، جستجو در جریان پردازش را برعکس می‌کند: queryها ذخیره می‌شوند و اسناد از کنار queryها می‌گذرند، مانند CEP. در ساده‌ترین حالت می‌توانید هر سند را با هر query آزمایش کنید، اگرچه با تعداد زیاد query کند می‌شود. برای بهینه‌سازی می‌توان queryها را هم ایندکس کرد و مجموعهٔ queryهای احتمالاً منطبق را محدود کرد [۷۷].

ارسال پیام و RPC

در «جریان داده با ارسال پیام» در صفحه ۱۳۶ سیستم‌های ارسال پیام را جایگزین RPC — مکانیزم ارتباط سرویس‌ها، مثلاً در مدل actor — بحث کردیم. اگرچه این سیستم‌ها هم بر پیام و رویداد استوارند، معمولاً پردازندهٔ جریانی نمی‌دانیم:

• چارچوب‌های actor عمدتاً مکانیزم مدیریت همزمانی و اجرای توزیع‌شدهٔ ماژول‌های ارتباطی‌اند، در حالی که پردازش جریانی عمدتاً تکنیک مدیریت داده است.
• ارتباط بین actorها اغلب گذرا و یک‌به‌یک است، در حالی که event log پایدار و چندمشترک است.
• actorها می‌توانند به هر شکل ارتباط برقرار کنند (از جمله الگوهای cyclic request/response)، اما پردازنده‌های جریانی معمولاً در pipelineهای acyclic تنظیم می‌شوند که هر جریان خروجی یک job مشخص و از مجموعهٔ مشخصی از جریان‌های ورودی مشتق است.

با این حال همپوشانی بین سیستم‌های شبیه RPC و پردازش جریانی وجود دارد. مثلاً Apache Storm ویژگی distributed RPC دارد که query کاربر را به گره‌هایی که جریان رویداد هم پردازش می‌کنند می‌فرستد؛ queryها با رویدادهای جریان‌های ورودی در هم می‌آیند و نتایج جمع و به کاربر برمی‌گردد [۷۸]. (به «پردازش داده چندپارتیشن» در صفحه ۵۱۴ هم مراجعه کنید.)

همچنین می‌توان جریان‌ها را با چارچوب‌های actor پردازش کرد. اما بسیاری از این چارچوب‌ها در crash تحویل پیام را تضمین نمی‌کنند، پس پردازش تحمل‌پذیر خطا نیست مگر منطق retry اضافی پیاده کنید.

استدلال دربارهٔ زمان

پردازنده‌های جریانی اغلب با زمان سر و کار دارند، به‌ویژه برای analytics که پنجره‌های زمانی مثل «میانگین پنج دقیقهٔ گذشته» به کار می‌رود. به نظر می‌رسد معنای «پنج دقیقهٔ گذشته» واضح باشد، اما مفهوم شگفت‌آور پیچیده است.

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

فرایند دسته‌ای ممکن است یک سال تاریخ را در چند دقیقه بخواند؛ در بیشتر موارد خط زمانی مورد علاقه همان سال تاریخ است، نه چند دقیقهٔ پردازش. علاوه بر این، استفاده از timestamp رویدادها پردازش را قطعی می‌کند: اجرای دوبارهٔ همان فرایند روی همان ورودی همان نتیجه را می‌دهد (به «تحمل خطا» در صفحه ۴۲۲ مراجعه کنید).

در مقابل، بسیاری از چارچوب‌های پردازش جریانی از ساعت محلی ماشین پردازش (processing time) برای windowing استفاده می‌کنند [۷۹]. این ساده است و اگر تأخیر بین ایجاد و پردازش رویداد ناچیز باشد منطقی است. اما اگر تأخیر پردازش قابل‌توجه باشد — یعنی پردازش به‌طور محسوس دیرتر از زمان واقعی رویداد رخ دهد — می‌شکند.

زمان رویداد در برابر زمان پردازش

دلایل زیادی برای تأخیر پردازش وجود دارد: صف‌بندی، خطای شبکه (به «شبکه‌های غیرقابل اعتماد» در صفحه ۲۷۷ مراجعه کنید)، مشکل کارایی و contention در broker یا پردازنده، restart مصرف‌کنندهٔ جریان یا پردازش مجدد رویدادهای گذشته (به «پخش مجدد پیام‌های قدیمی» در صفحه ۴۵۱ مراجعه کنید) هنگام بازیابی از خطا یا رفع باگ.

علاوه بر این، تأخیر پیام می‌تواند ترتیب غیرقابل پیش‌بینی پیام‌ها را ایجاد کند. مثلاً کاربر اول یک درخواست وب (سرور A) و بعد دومی (سرور B) می‌زند. A و B رویداد توصیف درخواست emit می‌کنند، اما رویداد B زودتر به broker می‌رسد. پردازندهٔ جریانی اول B و بعد A را می‌بیند، در حالی که واقعاً برعکس بوده. اگر تشبیه کمک کند، فیلم‌های جنگ ستارگان را در نظر بگیرید: قسمت IV در ۱۹۷۷، V در ۱۹۸۰، VI در ۱۹۸۳، سپس I، II و III در ۱۹۹۹، ۲۰۰۲ و ۲۰۰۵ و VII در ۲۰۱۵ [۸۰].ii اگر فیلم‌ها را به ترتیب اکران دیده باشید، ترتیب پردازش با ترتیب روایت ناسازگار است. (شمارهٔ قسمت مثل timestamp رویداد و تاریخ تماشا processing time است.) انسان‌ها با این ناپیوستگی‌ها کنار می‌آیند، اما الگوریتم‌های پردازش جریانی باید صریحاً برای مسائل زمان‌بندی و ترتیب نوشته شوند.

اشتباه گرفتن زمان رویداد و زمان پردازش به دادهٔ بد منجر می‌شود. مثلاً پردازندهٔ جریانی نرخ درخواست (تعداد در ثانیه) را اندازه می‌گیرد. اگر redeploy شود، شاید یک دقیقه خاموش و هنگام برگشت backlog را پردازش کند. اگر نرخ بر اساس processing time باشد، spike ناگهانی غیرعادی درخواست به نظر می‌رسد در حالی که نرخ واقعی ثابت بوده (شکل ۱۱-۷).

ii. سپاس از Kostas Kloudas از جامعهٔ Flink برای این تشبیه.

شکل ۱۱-۷. windowing بر اساس processing time به‌خاطر تغییرات نرخ پردازش artifact ایجاد می‌کند.

چه زمانی آماده‌اید؟

مسئلهٔ دشوار هنگام تعریف پنجره بر اساس زمان رویداد این است که هرگز مطمئن نیستید همهٔ رویدادهای یک پنجره را گرفته‌اید یا هنوز رویدادی می‌آید.

مثلاً رویدادها را در پنجره‌های یک‌دقیقه‌ای گروه کنید تا تعداد درخواست در دقیقه را بشمارید. تعدادی رویداد با timestamp دقیقهٔ ۳۷ام ساعت را شمرده‌اید و زمان جلو رفته؛ بیشتر رویدادهای ورودی در دقیقه‌های ۳۸ و ۳۹ام‌اند. چه زمانی پنجرهٔ دقیقهٔ ۳۷ را تمام اعلام و مقدار شمارنده را خروجی دهید؟

می‌توانید timeout کنید و پس از مدتی بدون رویداد جدید پنجره را آماده اعلام کنید، اما ممکن است رویدادهایی جایی buffer شده و به‌خاطر قطع شبکه تأخیر داشته باشند. باید stragglerهایی که پس از اعلام کامل پنجره می‌رسند را مدیریت کنید. به‌طور کلی دو گزینه دارید [۱]:

۱. stragglerها را نادیده بگیرید چون در شرایط عادی درصد کمی‌اند. تعداد رویدادهای drop‌شده را به‌عنوان متریک ردیابی و اگر مقدار قابل‌توجهی drop شد هشدار دهید.

۲. اصلاح publish کنید — مقدار به‌روز پنجره با stragglerها. شاید خروجی قبلی را retract هم کنید.

گاه می‌توان پیام ویژه‌ای فرستاد که «از این پس پیامی با timestamp زودتر از t نخواهد آمد» و مصرف‌کنندگان با آن پنجره را trigger کنند [۸۱]. اما اگر چند تولیدکننده روی ماشین‌های مختلف هر کدام آستانهٔ timestamp خود را دارند، مصرف‌کننده باید هر تولیدکننده را جدا ردیابی کند. افزودن و حذف تولیدکننده در این حالت سخت‌تر است.

اصلاً از ساعت چه کسی استفاده می‌کنید؟

تخصیص timestamp به رویداد وقتی رویداد در چند نقطه buffer می‌شود سخت‌تر است. اپ موبایل رویدادهای metrics را به سرور گزارش می‌دهد. ممکن است آفلاین استفاده شود و رویدادها را محلی buffer کند و هنگام اتصال بعدی (شاید ساعت‌ها یا روزها بعد) بفرستد. برای مصرف‌کنندگان جریان، رویدادها straggler بسیار تأخیردار به نظر می‌رسند. در این زمینه timestamp رویداد باید زمان تعامل کاربر طبق ساعت محلی دستگاه باشد. اما ساعت دستگاه کاربر قابل اعتماد نیست — ممکن است تصادفی یا عمدی اشتباه تنظیم شود (به «همگام‌سازی ساعت و دقت» در صفحه ۲۸۹ مراجعه کنید). زمان دریافت توسط سرور (طبق ساعت سرور) دقیق‌تر است چون سرور تحت کنترل شماست، اما برای توصیف تعامل کاربر کم‌معناست.

برای جبران ساعت نادرست دستگاه، یک رویکرد ثبت سه timestamp است [۸۲]:

• زمان وقوع رویداد طبق ساعت دستگاه
• زمان ارسال به سرور طبق ساعت دستگاه
• زمان دریافت توسط سرور طبق ساعت سرور

با تفریق دوم از سوم، offset بین ساعت دستگاه و سرور را تخمین می‌زنید (با فرض تأخیر شبکه ناچیز نسبت به دقت timestamp لازم). آن offset را روی timestamp رویداد اعمال می‌کنید و زمان واقعی وقوع را تخمین می‌زنید (با فرض offset ساعت دستگاه بین وقوع و ارسال تغییر نکرده).

این مسئله منحصر به پردازش جریانی نیست — پردازش دسته‌ای همان مسائل استدلال دربارهٔ زمان را دارد. فقط در زمینهٔ جریانی بیشتر از گذر زمان آگاهیم.

انواع پنجره

پس از تعیین timestamp رویداد، گام بعد تعریف پنجره روی بازه‌های زمانی است. پنجره برای تجمیع به کار می‌رود — مثلاً شمارش رویدادها یا میانگین مقادیر در پنجره. چند نوع پنجره رایج است [۷۹، ۸۳]:

Tumbling window پنجرهٔ tumbling طول ثابت دارد و هر رویداد دقیقاً به یک پنجره تعلق دارد. مثلاً با tumbling یک‌دقیقه‌ای، همهٔ رویدادهای بین ۱۰:۰۳:۰۰ و ۱۰:۰۳:۵۹ یک پنجره و بین ۱۰:۰۴:۰۰ و ۱۰:۰۴:۵۹ پنجرهٔ بعدی‌اند. می‌توانید با گرد کردن timestamp هر رویداد به نزدیک‌ترین دقیقه پنجره را تعیین کنید.

Hopping window پنجرهٔ hopping هم طول ثابت دارد اما اجازهٔ همپوشانی برای هموارسازی می‌دهد. مثلاً پنجرهٔ ۵ دقیقه با hop یک دقیقه رویدادهای بین ۱۰:۰۳:۰۰ و ۱۰:۰۷:۵۹ و پنجرهٔ بعدی بین ۱۰:۰۴:۰۰ و ۱۰:۰۸:۵۹ را پوشش می‌دهد. با محاسبهٔ tumbling یک‌دقیقه‌ای و تجمیع چند پنجرهٔ مجاور پیاده می‌شود.

Sliding window پنجرهٔ sliding همهٔ رویدادهایی را دارد که در فاصلهٔ مشخصی از یکدیگر رخ داده‌اند. مثلاً sliding ۵ دقیقه‌ای رویدادهای ۱۰:۰۳:۳۹ و ۱۰:۰۸:۱۲ را پوشش می‌دهد چون کمتر از ۵ دقیقه فاصله دارند (توجه: tumbling و hopping ۵ دقیقه‌ای با مرز ثابت این دو را در یک پنجره نمی‌گذارند). با نگه‌داشتن buffer رویدادها مرتب‌شده بر اساس زمان و حذف رویدادهای منقضی پیاده می‌شود.

Session window برخلاف انواع دیگر، session window طول ثابت ندارد. همهٔ رویدادهای یک کاربر که نزدیک هم در زمان رخ داده‌اند گروه می‌شوند و پنجره وقتی کاربر مدتی غیرفعال بود (مثلاً ۳۰ دقیقه بدون رویداد) پایان می‌یابد. sessionization نیاز رایج analytics وب‌سایت است (به «GROUP BY» در صفحه ۴۰۶ مراجعه کنید).

joinهای جریانی

در فصل ۱۰ دیدیم jobهای دسته‌ای چگونه مجموعه‌داده‌ها را بر اساس کلید join می‌کنند و joinها بخش مهمی از pipelineهای داده‌اند. چون پردازش جریانی pipeline داده را به پردازش تدریجی مجموعه‌داده‌های نامحدود تعمیم می‌دهد، همان نیاز به join روی جریان‌ها وجود دارد.

اما ظهور رویدادهای جدید هر لحظه روی جریان، join روی جریان را سخت‌تر از job دسته‌ای می‌کند. برای فهم بهتر، سه نوع join را جدا می‌کنیم: stream-stream، stream-table و table-table [۸۴]. در بخش‌های بعد هر کدام را با مثال نشان می‌دهیم.

join جریان-جریان (window join)

فرض کنید ویژگی جستجو در وب‌سایت دارید و می‌خواهید روندهای اخیر URLهای جستجو شده را تشخیص دهید. هر جستجو رویدادی با query و نتایج log می‌شود. هر کلیک روی نتیجه رویداد دیگری log می‌شود. برای محاسبهٔ click-through rate هر URL باید رویدادهای جستجو و کلیک را با همان session ID به هم وصل کنید. تحلیل‌های مشابه در سیستم‌های تبلیغاتی لازم است [۸۵].

کلیک ممکن است هرگز نیاید اگر کاربر جستجو را رها کند، و حتی اگر بیاید فاصلهٔ جستجو تا کلیک بسیار متغیر است: اغلب چند ثانیه، اما تا روزها یا هفته‌ها (کاربر جستجو می‌کند، تب را فراموش و بعد برمی‌گردد). به‌خاطر تأخیر شبکه، رویداد کلیک حتی ممکن است قبل از رویداد جستجو برسد. پنجرهٔ مناسب برای join انتخاب کنید — مثلاً کلیک را با جستجو join کنید اگر حداکثر یک ساعت فاصله داشته باشند.

توجه: embed کردن جزئیات جستجو در رویداد کلیک معادل join نیست: فقط مواردی را می‌بینید که کاربر کلیک کرده، نه جستجوهایی که کلیکی نداشته‌اند. برای سنجش کیفیت جستجو به click-through rate دقیق نیاز دارید و به هر دو رویداد جستجو و کلیک.

برای این join، پردازندهٔ جریانی باید وضعیت نگه دارد — مثلاً همهٔ رویدادهای یک ساعت گذشته ایندکس‌شده با session ID. با هر رویداد جستجو یا کلیک، به ایندکس مناسب اضافه و ایندکس دیگر برای رویداد منطبق همان session ID بررسی می‌شود. با تطابق، رویدادی emit می‌کنید که کدام نتیجهٔ جستجو کلیک شده. اگر رویداد جستجو بدون کلیک منطبق منقضی شود، رویدادی emit می‌کنید که کدام نتایج کلیک نشده‌اند.

join جریان-جدول (غنی‌سازی جریان)

در «مثال: تحلیل رویدادهای فعالیت کاربر» در صفحه ۴۰۴ (شکل ۱۰-۲) job دسته‌ای دو مجموعه‌داده را join می‌کرد: رویدادهای فعالیت کاربر و پایگاه پروفایل کاربر. طبیعی است رویدادهای فعالیت را جریان بدانید و همان join را پیوسته در پردازندهٔ جریانی انجام دهید: ورودی جریان رویدادهای فعالیت با user ID است و خروجی جریان رویدادهای فعالیت غنی‌شده با اطلاعات پروفایل است. این فرایند گاه غنی‌سازی (enrichment) رویدادهای فعالیت با اطلاعات پایگاه داده نامیده می‌شود.

برای این join، پردازنده یک رویداد فعالیت را می‌بیند، user ID را در پایگاه داده جستجو و اطلاعات پروفایل را اضافه می‌کند. جستجوی پایگاه داده می‌تواند query از راه دور باشد؛ اما همان‌طور که در «مثال: تحلیل رویدادهای فعالیت کاربر» در صفحه ۴۰۴ بحث شد، query از راه دور کند است و خطر overload پایگاه داده دارد [۷۵].

رویکرد دیگر بارگذاری کپی پایگاه داده در پردازندهٔ جریانی برای query محلی بدون round-trip شبکه است. بسیار شبیه hash joinهای «joinهای سمت map» در صفحه ۴۰۸: کپی محلی ممکن است hash table در حافظه اگر کوچک باشد یا ایندکس روی دیسک محلی.

تفاوت با job دسته‌ای این است که job دسته‌ای از snapshot لحظه‌ای پایگاه داده به‌عنوان ورودی استفاده می‌کند، در حالی که پردازندهٔ جریانی long-running است و محتوای پایگاه داده در طول زمان تغییر می‌کند، پس کپی محلی باید به‌روز بماند. این با CDC حل می‌شود: پردازنده علاوه بر جریان رویدادهای فعالیت، به changelog پایگاه پروفایل کاربر subscribe می‌کند. با ایجاد یا تغییر پروفایل، کپی محلی به‌روز می‌شود. پس join بین دو جریان داریم: رویدادهای فعالیت و به‌روزرسانی‌های پروفایل.

join جریان-جدول بسیار شبیه join جریان-جریان است؛ بزرگ‌ترین تفاوت این است که برای جریان changelog جدول، join از پنجره‌ای تا «ابتدای زمان» (پنجرهٔ مفهومی نامحدود) استفاده می‌کند و نسخه‌های جدیدتر رکوردها قدیمی‌تر را overwrite می‌کنند. برای ورودی جریان، join شاید اصلاً پنجره نگه ندارد.

join جدول-جدول (نگه‌داری نمای materialized)

مثال timeline توییتر را در «توصیف بار» در صفحه ۱۱ در نظر بگیرید. گفتیم برای نمایش timeline خانه، iterate روی همهٔ افراد دنبال‌شده و جمع توییت‌های اخیرشان و ادغام گران است.

به cache timeline نیاز داریم: نوعی «صندوق ورودی» per-user که توییت‌ها هنگام ارسال در آن نوشته می‌شوند تا خواندن timeline یک lookup باشد. materialize و نگه‌داری این cache به پردازش رویداد زیر نیاز دارد:

• وقتی کاربر u توییت جدید می‌فرستد، به timeline همهٔ دنبال‌کنندگان u اضافه می‌شود.
• وقتی کاربر توییت حذف می‌کند، از همهٔ timelineها حذف می‌شود.
• وقتی u1 شروع به دنبال کردن u2 می‌کند، توییت‌های اخیر u2 به timeline u1 اضافه می‌شود.
• وقتی u1 unfollow می‌کند، توییت‌های u2 از timeline u1 حذف می‌شود.

برای نگه‌داری cache در پردازندهٔ جریانی به جریان‌های رویداد توییت (ارسال و حذف) و روابط follow (follow و unfollow) نیاز دارید. پردازندهٔ جریانی پایگاهی از مجموعهٔ follower هر کاربر نگه می‌دارد تا بداند هنگام توییت جدید کدام timelineها به‌روز شوند [۸۶].

نگاه دیگر: پردازنده materialized view برای queryیی که دو جدول (tweets و follows) را join می‌کند نگه می‌دارد، چیزی شبیه:

SELECT follows.follower_id AS timeline_id,
  array_agg(tweets.* ORDER BY tweets.timestamp DESC)
FROM tweets
JOIN follows ON follows.followee_id = tweets.sender_id
GROUP BY follows.follower_id

join جریان‌ها مستقیماً با join جداول در آن query مطابقت دارد. timelineها عملاً cache نتیجهٔ این query هستند که با هر تغییر جداول زیرین به‌روز می‌شوند.iii

iii. اگر جریان را مشتق جدول بدانید (شکل ۱۱-۶) و join را حاصل ضرب دو جدول u·v، اتفاق جالبی می‌افتد: جریان تغییرات join materialized از قاعدهٔ حاصلضرب (u·v)′ = u′v + uv′ پیروی می‌کند. به زبان ساده: هر تغییر توییت با followerهای فعلی join می‌شود و هر تغییر follower با توییت‌های فعلی [۴۹، ۵۰].

وابستگی زمانی joinها

سه نوع join (stream-stream، stream-table، table-table) شباهت زیادی دارند: همه به پردازندهٔ جریانی نیاز دارند وضعیت را بر اساس یک ورودی join نگه دارد (رویدادهای جستجو و کلیک، پروفایل کاربر یا لیست follower) و هنگام پیام‌های ورودی دیگر آن وضعیت را query کند.

ترتیب رویدادهایی که وضعیت را نگه می‌دارند مهم است (اول follow و بعد unfollow مهم است یا برعکس). در لاگ پارتیشن‌بندی‌شده ترتیب رویدادها در یک پارتیشن حفظ می‌شود، اما معمولاً تضمین ترتیب بین جریان‌ها یا پارتیشن‌های مختلف نیست.

این سؤال را مطرح می‌کند: اگر رویدادها روی جریان‌های مختلف تقریباً همزمان باشند، به چه ترتیبی پردازش می‌شوند؟ در join جریان-جدول، اگر کاربر پروفایل را به‌روز کند، کدام رویدادهای فعالیت با پروفایل قدیمی (قبل از به‌روزرسانی) و کدام با جدید join می‌شوند؟ به عبارت دیگر: اگر وضعیت در زمان تغییر کند و با وضعیتی join کنید، از کدام نقطهٔ زمانی برای join استفاده می‌کنید [۴۵]؟

چنین وابستگی زمانی در جاهای زیاد رخ می‌دهد. مثلاً اگر چیزی می‌فروشید، باید نرخ مالیات درست را روی فاکتور اعمال کنید که به کشور یا ایالت، نوع محصول و تاریخ فروش بستگی دارد (نرخ مالیات گاه تغییر می‌کند). هنگام join فروش با جدول نرخ مالیات، می‌خواهید با نرخ زمان فروش join کنید که ممکن است با نرخ فعلی هنگام پردازش مجدد دادهٔ تاریخی متفاوت باشد.

اگر ترتیب رویدادها بین جریان‌ها نامشخص باشد، join غیرقطعی می‌شود [۸۷] — یعنی نمی‌توانید همان job را روی همان ورودی دوباره اجرا کنید و حتماً همان نتیجه را بگیرید: رویدادهای ورودی ممکن است دفعهٔ دیگر به شکل دیگری در هم آمیخته شوند.

در انبار داده این مسئله slowly changing dimension (SCD) نامیده می‌شود و اغلب با شناسهٔ یکتا برای نسخهٔ خاص رکورد join‌شده حل می‌شود: مثلاً هر بار تغییر نرخ مالیات شناسهٔ جدید می‌گیرد و فاکتور شناسهٔ نرخ زمان فروش را دارد [۸۸، ۸۹]. این join را قطعی می‌کند، اما log compaction ممکن نیست چون همهٔ نسخه‌های رکورد در جدول باید بمانند.

تحمل خطا

در بخش پایانی فصل ببینیم پردازنده‌های جریانی چگونه خطا را تحمل می‌کنند. در فصل ۱۰ دیدیم چارچوب‌های پردازش دسته‌ای نسبتاً راحت خطا را تحمل می‌کنند: اگر task در job MapReduce شکست بخورد، روی ماشین دیگر شروع می‌شود و خروجی task شکست‌خورده دور ریخته می‌شود. این retry شفاف ممکن است چون فایل‌های ورودی تغییرناپذیرند، هر task خروجی را در فایل جدا روی HDFS می‌نویسد و خروجی فقط با موفقیت task visible می‌شود.

رویکرد دسته‌ای به تحمل خطا تضمین می‌کند خروجی job همان‌طور است که گویی هیچ مشکلی نبوده، حتی اگر برخی taskها شکست خورده باشند. گویی هر رکورد ورودی دقیقاً یک‌بار پردازش شده — هیچ رکوردی رد نشده و دوبار پردازش نشده. با وجود اینکه restart taskها ممکن است رکوردها را چند بار پردازش کند، اثر visible در خروجی گویی یک‌بار پردازش شده است. این اصل exactly-once semantics نامیده می‌شود، اگرچه effectively-once توصیفی‌تر است [۹۰].

همان مسئلهٔ تحمل خطا در پردازش جریانی هم هست، اما سخت‌تر: منتظر ماندن تا task تمام شود قبل از visible کردن خروجی گزینه نیست چون جریان نامحدود است و هرگز پردازش تمام نمی‌شود.

microbatch و checkpoint

یک راه‌حل شکستن جریان به بلوک‌های کوچک و رفتار با هر بلوک مثل فرایند دسته‌ای کوچک است. این microbatching نامیده می‌شود و در Spark Streaming [۹۱] به کار می‌رود. اندازهٔ batch معمولاً حدود یک ثانیه است — trade-off کارایی: batch کوچک‌تر سربار scheduling و هماهنگی بیشتر، batch بزرگ‌تر تأخیر بیشتر قبل از visible شدن نتایج.

microbatching به‌طور ضمنی tumbling window به اندازهٔ batch (بر اساس processing time نه timestamp رویداد) می‌دهد؛ jobهایی که پنجرهٔ بزرگ‌تر می‌خواهند باید صریحاً وضعیت را از microbatch به بعدی منتقل کنند.

رویکرد متغیر در Apache Flink، تولید دوره‌ای checkpointهای rolling از وضعیت و نوشتن روی ذخیرهٔ پایدار است [۹۲، ۹۳]. اگر operator جریان crash کند، از آخرین checkpoint restart و خروجی تولیدشده بین checkpoint و crash دور ریخته می‌شود. checkpointها با barrier در جریان پیام trigger می‌شوند، شبیه مرز بین microbatchها اما بدون اجبار اندازهٔ پنجرهٔ خاص.

در محدودهٔ چارچوب پردازش جریانی، microbatching و checkpoint همان semantics exactly-once پردازش دسته‌ای را می‌دهند. اما به‌محض خروج از پردازندهٔ جریان (مثلاً نوشتن به پایگاه داده، ارسال به broker خارجی یا ایمیل)، چارچوب دیگر نمی‌تواند خروجی batch شکست‌خورده را دور بیندازد. در این حالت restart task شکست‌خورده باعث تکرار اثر جانبی خارجی می‌شود و microbatching یا checkpoint به‌تنهایی کافی نیست.

بازبینی commit اتمی

برای ظاهر exactly-once در حضور خطا باید اطمینان حاصل شود همهٔ خروجی‌ها و اثرات جانبی پردازش رویداد فقط و فقط در صورت موفقیت پردازش اثر کنند. آن اثرات شامل پیام به operatorهای downstream یا سیستم‌های پیام‌رسانی خارجی (از جمله ایمیل یا push)، write به پایگاه داده، تغییر وضعیت operator و acknowledgment پیام‌های ورودی (از جمله جلو بردن offset مصرف‌کننده در broker مبتنی بر لاگ) است.

این‌ها یا همه باید اتمی رخ دهند یا هیچ‌کدام — نباید با یکدیگر out of sync شوند. اگر آشناست، چون در «پردازش پیام exactly-once» در صفحه ۳۶۰ در زمینهٔ تراکنش‌های توزیع‌شده و two-phase commit بحث شد.

در فصل ۹ مشکلات پیاده‌سازی‌های سنتی تراکنش‌های توزیع‌شده مثل XA را دیدیم. اما در محیط‌های محدودتر می‌توان چنین commit اتمی را کارآمد پیاده کرد. این رویکرد در Google Cloud Dataflow [۸۱، ۹۲] و VoltDB [۹۴] به کار می‌رود و برنامه‌هایی برای افزودن ویژگی مشابه به Apache Kafka وجود دارد [۹۵، ۹۶]. برخلاف XA، این پیاده‌سازی‌ها تراکنش بین فناوری‌های ناهمگن نمی‌خواهند، بلکه هر دو را درون چارچوب پردازش جریانی مدیریت می‌کنند. سربار پروتکل تراکنش با پردازش چند پیام ورودی در یک تراکنش amortize می‌شود.

idempotence

هدف دور ریختن خروجی جزئی taskهای شکست‌خورده است تا بتوانند بدون اثر دوبار امن retry شوند. تراکنش‌های توزیع‌شده یک راه است، راه دیگر تکیه بر idempotence [۹۷].

عملی idempotent عملی است که چند بار انجام دهید همان اثر یک‌بار انجام دادن را دارد. مثلاً تنظیم کلید در key-value store به مقدار ثابت idempotent است (نوشتن دوباره همان مقدار را overwrite می‌کند)، اما increment شمارنده idempotent نیست (دوبار increment یعنی دو برابر افزایش).

حتی اگر عملی ذاتاً idempotent نباشد، اغلب با کمی metadata idempotent می‌شود. مثلاً هنگام مصرف پیام از Kafka، هر پیام offset پایدار monotonic افزایشی دارد. هنگام نوشتن به پایگاه خارجی می‌توانید offset پیامی که آخرین write را trigger کرد با مقدار بنویسید. پس می‌فهمید به‌روزرسانی قبلاً اعمال شده و از تکرار اجتناب می‌کنید.

مدیریت وضعیت در Trident استورم بر ایدهٔ مشابهی است [۷۸]. تکیه بر idempotence چند فرض دارد: restart task شکست‌خورده باید همان پیام‌ها را به همان ترتیب replay کند (broker مبتنی بر لاگ این را می‌دهد)، پردازش باید قطعی باشد و گره دیگری همزمان همان مقدار را به‌روز نکند [۹۸، ۹۹].

هنگام failover از یک گره پردازش به دیگر، شاید fencing لازم باشد (به «leader و قفل» در صفحه ۳۰۱ مراجعه کنید) تا تداخل گره‌ای که مرده فرض شده اما زنده است جلوگیری شود. با وجود این محدودیت‌ها، عملیات idempotent راه مؤثری برای exactly-once با سربار کم است.

بازسازی وضعیت پس از خرابی

هر پردازش جریانی که به وضعیت نیاز دارد — مثلاً تجمیع‌های پنجره‌ای (شمارنده، میانگین، histogram) و جداول و ایندکس‌های join — باید اطمینان دهد وضعیت پس از خرابی قابل بازیابی است.

یک گزینه نگه‌داری وضعیت در datastore راه‌دور و replicate آن است، اگرچه query پایگاه راه‌دور برای هر پیام کند است (به «join جریان-جدول» در صفحه ۴۷۳ مراجعه کنید). جایگزین نگه‌داری وضعیت محلی در پردازندهٔ جریانی و replicate دوره‌ای است. هنگام بازیابی، task جدید وضعیت replicate‌شده را می‌خواند و بدون از دست دادن داده ادامه می‌دهد.

مثلاً Flink دوره‌ای snapshot از وضعیت operator می‌گیرد و روی HDFS می‌نویسد [۹۲، ۹۳]؛ Samza و Kafka Streams تغییرات وضعیت را با ارسال به topic اختصاصی Kafka با log compaction replicate می‌کنند، شبیه CDC [۸۴، ۱۰۰]. VoltDB با پردازش redundant هر پیام ورودی روی چند گره وضعیت را replicate می‌کند (به «اجرای واقعاً ترتیبی» در صفحه ۲۵۲ مراجعه کنید).

گاه replicate وضعیت لازم نیست چون از جریان‌های ورودی قابل بازسازی است. اگر وضعیت تجمیع روی پنجرهٔ کوتاه باشد، شاید replay رویدادهای ورودی آن پنجره به‌اندازهٔ کافی سریع باشد. اگر وضعیت کپی محلی پایگاه داده با CDC نگه داشته می‌شود، پایگاه داده را هم از جریان تغییر فشرده‌شدهٔ لاگ می‌توان بازسازی کرد (به «فشرده‌سازی لاگ» در صفحه ۴۵۶ مراجعه کنید).

با این حال همهٔ این trade-offها به ویژگی‌های کارایی زیرساخت بستگی دارد: در برخی سیستم‌ها تأخیر شبکه از latency دسترسی دیسک کمتر و پهنای باند شبکه با دیسک قابل مقایسه است. trade-off ایده‌آل جهانی برای همهٔ موقعیت‌ها نیست و مزایای وضعیت محلی در برابر راه‌دور با تکامل فناوری ذخیره و شبکه هم تغییر می‌کند.

خلاصه

در این فصل جریان‌های رویداد، کاربردها و نحوهٔ پردازش آن‌ها را بحث کردیم. از برخی جهات، پردازش جریانی بسیار شبیه پردازش دسته‌ای فصل ۱۰ است، اما به‌صورت پیوسته روی جریان‌های نامحدود (هرگز پایان‌نیافته) به‌جای ورودی با اندازهٔ ثابت. از این منظر، message brokerها و event log معادل جریانی فایل‌سیستم‌اند.

زمانی را صرف مقایسهٔ دو نوع message broker کردیم:

Message broker سبک AMQP/JMS broker پیام‌های تکی را به مصرف‌کنندگان تخصیص می‌دهد و مصرف‌کنندگان پس از پردازش موفق پیام‌های تکی را تأیید می‌کنند. پیام‌ها پس از تأیید از broker حذف می‌شوند. این رویکرد به‌عنوان شکل ناهمگام RPC مناسب است (به «جریان داده با ارسال پیام» در صفحه ۱۳۶ هم مراجعه کنید)، مثلاً در صف کار، جایی که ترتیب دقیق پردازش پیام مهم نیست و نیازی به بازخوانی پیام‌های قدیمی پس از پردازش نیست.

Message broker مبتنی بر لاگ broker همهٔ پیام‌های یک پارتیشن را به همان گره مصرف‌کننده تخصیص می‌دهد و همیشه پیام‌ها را به همان ترتیب تحویل می‌دهد. موازی‌سازی با پارتیشن‌بندی است و مصرف‌کنندگان با checkpoint کردن offset آخرین پیام پردازش‌شده پیشرفت را ردیابی می‌کنند. broker پیام‌ها را روی دیسک نگه می‌دارد، پس می‌توان برای خواندن مجدد پیام‌های قدیمی برگشت.

رویکرد مبتنی بر لاگ شباهت‌هایی با replication log پایگاه‌های داده (فصل ۵) و موتورهای log-structured (فصل ۳) دارد. دیدیم این رویکرد به‌ویژه برای سیستم‌های پردازش جریانی که جریان‌های ورودی را مصرف و وضعیت یا جریان‌های خروجی مشتق‌شده تولید می‌کنند مناسب است.

دربارهٔ منشأ جریان‌ها چند امکان را دیدیم: رویدادهای فعالیت کاربر، خوانش‌های دوره‌ای سنسورها و feed داده (مثلاً دادهٔ بازار در مالی) به‌طور طبیعی جریان‌اند. دیدیم مفید است write به پایگاه داده را هم جریان بدانید: می‌توانید changelog — تاریخچهٔ همهٔ تغییرات پایگاه داده — را ضمناً با CDC یا صریحاً با event sourcing capture کنید. log compaction به جریان اجازه می‌دهد کپی کامل محتوای پایگاه داده را نگه دارد.

نمایش پایگاه‌های داده به‌صورت جریان‌ها فرصت‌های قدرتمندی برای یکپارچه‌سازی سیستم‌ها باز می‌کند. می‌توانید سیستم‌های مشتق‌شده مثل ایندکس جستجو، cache و analytics را با مصرف لاگ تغییرات و اعمال آن‌ها به‌روز نگه دارید. حتی می‌توانید با شروع از صفر و مصرف لاگ تغییرات از ابتدا تا امروز، نماهای تازه روی دادهٔ موجود بسازید.

امکانات نگه‌داری وضعیت به‌صورت جریان و replay پیام‌ها هم پایهٔ تکنیک‌هایی است که join جریانی و تحمل خطا را در چارچوب‌های پردازش جریانی ممکن می‌کند. چند هدف پردازش جریانی را بحث کردیم: جستجوی الگوهای رویداد (CEP)، محاسبهٔ تجمیع‌های پنجره‌ای (تحلیل جریانی) و به‌روز نگه‌داشتن سیستم‌های دادهٔ مشتق‌شده (materialized view).

سپس دشواری‌های استدلال دربارهٔ زمان در پردازندهٔ جریانی را بحث کردیم: تفاوت زمان پردازش و timestamp رویداد، و مسئلهٔ stragglerهایی که پس از اعلام کامل بودن پنجره می‌رسند.

سه نوع join در فرایندهای جریانی را تمایز دادیم:

Join جریان-جریان هر دو ورودی جریان رویدادهای فعالیت‌اند و operator join رویدادهای مرتبط را در بازهٔ زمانی می‌جوید. مثلاً دو عمل یک کاربر در ۳۰ دقیقهٔ یکدیگر را match می‌کند. دو ورودی join ممکن است همان جریان باشد (self-join) اگر رویدادهای مرتبط در همان جریان را بخواهید.

Join جریان-جدول یک ورودی جریان رویدادهای فعالیت و دیگری changelog پایگاه داده است. changelog کپی محلی پایگاه داده را به‌روز نگه می‌دارد. برای هر رویداد فعالیت، operator پایگاه داده را query و رویداد فعالیت غنی‌شده خروجی می‌دهد.

Join جدول-جدول هر دو ورودی changelog پایگاه داده‌اند. در این حالت هر تغییر در یک طرف با آخرین وضعیت طرف دیگر join می‌شود. نتیجه جریان تغییرات materialized view از join دو جدول است.

در پایان تکنیک‌های دستیابی به تحمل خطا و semantics exactly-once در پردازندهٔ جریانی را بحث کردیم. مانند پردازش دسته‌ای، باید خروجی جزئی taskهای شکست‌خورده را دور بیندازیم. اما چون فرایند جریانی long-running است و پیوسته خروجی می‌دهد، نمی‌توان همهٔ خروجی را ساده دور ریخت. در عوض مکانیزم بازیابی ریزدانه‌تری بر پایهٔ microbatching، checkpoint، تراکنش یا writeهای idempotent به کار می‌رود.

منابع

[1] Tyler Akidau, Robert Bradshaw, Craig Chambers, et al.: "The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing," Proceedings of the VLDB Endowment, volume 8, number 12, pages 1792–1803, August 2015. doi:10.14778/2824032.2824076

[2] Harold Abelson, Gerald Jay Sussman, and Julie Sussman: Structure and Interpretation of Computer Programs, 2nd edition. MIT Press, 1996. ISBN: 978-0-262-51087-5, available online at mitpress.mit.edu

[3] Patrick Th. Eugster, Pascal A. Felber, Rachid Guerraoui, and Anne-Marie Kermarrec: "The Many Faces of Publish/Subscribe," ACM Computing Surveys, volume 35, number 2, pages 114–131, June 2003. doi:10.1145/857076.857078

[4] Joseph M. Hellerstein and Michael Stonebraker: Readings in Database Systems, 4th edition. MIT Press, 2005. ISBN: 978-0-262-69314-1, available online at redbook.cs.berkeley.edu

[5] Don Carney, Uğur Çetintemel, Mitch Cherniack, et al.: "Monitoring Streams – A New Class of Data Management Applications," at 28th International Conference on Very Large Data Bases (VLDB), August 2002.

[6] Matthew Sackman: "Pushing Back," lshift.net, May 5, 2016.

[7] Vicent Martí: "Brubeck, a statsd-Compatible Metrics Aggregator," githubengineering.com, June 15, 2015.

[8] Seth Lowenberger: "MoldUDP64 Protocol Specification V 1.00," nasdaqtrader.com, July 2009.

[9] Pieter Hintjens: ZeroMQ – The Guide. O'Reilly Media, 2013. ISBN: 978-1-449-33404-8

[10] Ian Malpass: "Measure Anything, Measure Everything," codeascraft.com, February 15, 2011.

[11] Dieter Plaetinck: "25 Graphite, Grafana and statsd Gotchas," blog.raintank.io, March 3, 2016.

[12] Jeff Lindsay: "Web Hooks to Revolutionize the Web," progrium.com, May 3, 2007.

[13] Jim N. Gray: "Queues Are Databases," Microsoft Research Technical Report MSR-TR-95-56, December 1995.

[14] Mark Hapner, Rich Burridge, Rahul Sharma, et al.: "JSR-343 Java Message Service (JMS) 2.0 Specification," jms-spec.java.net, March 2013.

[15] Sanjay Aiyagari, Matthew Arrott, Mark Atwell, et al.: "AMQP: Advanced Message Queuing Protocol Specification," Version 0-9-1, November 2008.

[16] "Google Cloud Pub/Sub: A Google-Scale Messaging Service," cloud.google.com, 2016.

[17] "Apache Kafka 0.9 Documentation," kafka.apache.org, November 2015.

[18] Jay Kreps, Neha Narkhede, and Jun Rao: "Kafka: A Distributed Messaging System for Log Processing," at 6th International Workshop on Networking Meets Databases (NetDB), June 2011.

[19] "Amazon Kinesis Streams Developer Guide," docs.aws.amazon.com, April 2016.

[20] Leigh Stewart and Sijie Guo: "Building DistributedLog: Twitter's High-Performance Replicated Log Service," blog.twitter.com, September 16, 2015.

[21] "DistributedLog Documentation," Twitter, Inc., distributedlog.io, May 2016.

[22] Jay Kreps: "Benchmarking Apache Kafka: 2 Million Writes Per Second (On Three Cheap Machines)," engineering.linkedin.com, April 27, 2014.

[23] Kartik Paramasivam: "How We're Improving and Advancing Kafka at LinkedIn," engineering.linkedin.com, September 2, 2015.

[24] Jay Kreps: "The Log: What Every Software Engineer Should Know About Real-Time Data's Unifying Abstraction," engineering.linkedin.com, December 16, 2013.

[25] Shirshanka Das, Chavdar Botev, Kapil Surlaker, et al.: "All Aboard the Databus!," at 3rd ACM Symposium on Cloud Computing (SoCC), October 2012.

[26] Yogeshwer Sharma, Philippe Ajoux, Petchean Ang, et al.: "Wormhole: Reliable Pub-Sub to Support Geo-Replicated Internet Services," at 12th USENIX Symposium on Networked Systems Design and Implementation (NSDI), May 2015.

[27] P. P. S. Narayan: "Sherpa Update," developer.yahoo.com, June 8, 2014.

[28] Martin Kleppmann: "Bottled Water: Real-Time Integration of PostgreSQL and Kafka," martin.kleppmann.com, April 23, 2015.

[29] Ben Osheroff: "Introducing Maxwell, a mysql-to-kafka Binlog Processor," developer.zendesk.com, August 20, 2015.

[30] Randall Hauch: "Debezium 0.2.1 Released," debezium.io, June 10, 2016.

[31] Prem Santosh Udaya Shankar: "Streaming MySQL Tables in Real-Time to Kafka," engineeringblog.yelp.com, August 1, 2016.

[32] "Mongoriver," Stripe, Inc., github.com, September 2014.

[33] Dan Harvey: "Change Data Capture with Mongo + Kafka," at Hadoop Users Group UK, August 2015.

[34] "Oracle GoldenGate 12c: Real-Time Access to Real-Time Information," Oracle White Paper, March 2015.

[35] "Oracle GoldenGate Fundamentals: How Oracle GoldenGate Works," Oracle Corporation, youtube.com, November 2012.

[36] Slava Akhmechet: "Advancing the Realtime Web," rethinkdb.com, January 27, 2015.

[37] "Firebase Realtime Database Documentation," Google, Inc., firebase.google.com, May 2016.

[38] "Apache CouchDB 1.6 Documentation," docs.couchdb.org, 2014.

[39] Matt DeBergalis: "Meteor 0.7.0: Scalable Database Queries Using MongoDB Oplog Instead of Poll-and-Diff," info.meteor.com, December 17, 2013.

[40] "Chapter 15. Importing and Exporting Live Data," VoltDB 6.4 User Manual, docs.voltdb.com, June 2016.

[41] Neha Narkhede: "Announcing Kafka Connect: Building Large-Scale Low-Latency Data Pipelines," confluent.io, February 18, 2016.

[42] Greg Young: "CQRS and Event Sourcing," at Code on the Beach, August 2014.

[43] Martin Fowler: "Event Sourcing," martinfowler.com, December 12, 2005.

[44] Vaughn Vernon: Implementing Domain-Driven Design. Addison-Wesley Professional, 2013. ISBN: 978-0-321-83457-7

[45] H. V. Jagadish, Inderpal Singh Mumick, and Abraham Silberschatz: "View Maintenance Issues for the Chronicle Data Model," at 14th ACM SIGACT-SIGMOD-SIGART Symposium on Principles of Database Systems (PODS), May 1995. doi:10.1145/212433.220201

[46] "Event Store 3.5.0 Documentation," Event Store LLP, docs.geteventstore.com, February 2016.

[47] Martin Kleppmann: Making Sense of Stream Processing. Report, O'Reilly Media, May 2016.

[48] Sander Mak: "Event-Sourced Architectures with Akka," at JavaOne, September 2014.

[49] Julian Hyde: personal communication, June 2016.

[50] Ashish Gupta and Inderpal Singh Mumick: Materialized Views: Techniques, Implementations, and Applications. MIT Press, 1999. ISBN: 978-0-262-57122-7

[51] Timothy Griffin and Leonid Libkin: "Incremental Maintenance of Views with Duplicates," at ACM International Conference on Management of Data (SIGMOD), May 1995. doi:10.1145/223784.223849

[52] Pat Helland: "Immutability Changes Everything," at 7th Biennial Conference on Innovative Data Systems Research (CIDR), January 2015.

[53] Martin Kleppmann: "Accounting for Computer Scientists," martin.kleppmann.com, March 7, 2011.

[54] Pat Helland: "Accountants Don't Use Erasers," blogs.msdn.com, June 14, 2007.

[55] Fangjin Yang: "Dogfooding with Druid, Samza, and Kafka: Metametrics at Metamarkets," metamarkets.com, June 3, 2015.

[56] Gavin Li, Jianqiu Lv, and Hang Qi: "Pistachio: Co-Locate the Data and Compute for Fastest Cloud Compute," yahoohadoop.tumblr.com, April 13, 2015.

[57] Kartik Paramasivam: "Stream Processing Hard Problems – Part 1: Killing Lambda," engineering.linkedin.com, June 27, 2016.

[58] Martin Fowler: "CQRS," martinfowler.com, July 14, 2011.

[59] Greg Young: "CQRS Documents," cqrs.files.wordpress.com, November 2010.

[60] Baron Schwartz: "Immutability, MVCC, and Garbage Collection," xaprb.com, December 28, 2013.

[61] Daniel Eloff, Slava Akhmechet, Jay Kreps, et al.: "Re: Turning the Database Inside-out with Apache Samza," Hacker News discussion, news.ycombinator.com, March 4, 2015.

[62] "Datomic Development Resources: Excision," Cognitect, Inc., docs.datomic.com.

[63] "Fossil Documentation: Deleting Content from Fossil," fossil-scm.org, 2016.

[64] Jay Kreps: "The irony of distributed systems is that data loss is really easy but deleting data is surprisingly hard," twitter.com, March 30, 2015.

[65] David C. Luckham: "What's the Difference Between ESP and CEP?," complexevents.com, August 1, 2006.

[66] Srinath Perera: "How Is Stream Processing and Complex Event Processing (CEP) Different?," quora.com, December 3, 2015.

[67] Arvind Arasu, Shivnath Babu, and Jennifer Widom: "The CQL Continuous Query Language: Semantic Foundations and Query Execution," The VLDB Journal, volume 15, number 2, pages 121–142, June 2006. doi:10.1007/s00778-004-0147-z

[68] Julian Hyde: "Data in Flight: How Streaming SQL Technology Can Help Solve the Web 2.0 Data Crunch," ACM Queue, volume 7, number 11, December 2009. doi:10.1145/1661785.1667562

[69] "Esper Reference, Version 5.4.0," EsperTech, Inc., espertech.com, April 2016.

[70] Zubair Nabi, Eric Bouillet, Andrew Bainbridge, and Chris Thomas: "Of Streams and Storms," IBM technical report, developer.ibm.com, April 2014.

[71] Milinda Pathirage, Julian Hyde, Yi Pan, and Beth Plale: "SamzaSQL: Scalable Fast Data Management with Streaming SQL," at IEEE International Workshop on High-Performance Big Data Computing (HPBDC), May 2016. doi:10.1109/IPDPSW.2016.141

[72] Philippe Flajolet, Éric Fusy, Olivier Gandouet, and Frédéric Meunier: "HyperLogLog: The Analysis of a Near-Optimal Cardinality Estimation Algorithm," at Conference on Analysis of Algorithms (AofA), June 2007.

[73] Jay Kreps: "Questioning the Lambda Architecture," oreilly.com, July 2, 2014.

[74] Ian Hellström: "An Overview of Apache Streaming Technologies," databaseline.wordpress.com, March 12, 2016.

[75] Jay Kreps: "Why Local State Is a Fundamental Primitive in Stream Processing," oreilly.com, July 31, 2014.

[76] Shay Banon: "Percolator," elastic.co, February 8, 2011.

[77] Alan Woodward and Martin Kleppmann: "Real-Time Full-Text Search with Luwak and Samza," martin.kleppmann.com, April 13, 2015.

[78] "Apache Storm 1.0.1 Documentation," storm.apache.org, May 2016.

[79] Tyler Akidau: "The World Beyond Batch: Streaming 102," oreilly.com, January 20, 2016.

[80] Stephan Ewen: "Streaming Analytics with Apache Flink," at Kafka Summit, April 2016.

[81] Tyler Akidau, Alex Balikov, Kaya Bekiroğlu, et al.: "MillWheel: Fault-Tolerant Stream Processing at Internet Scale," at 39th International Conference on Very Large Data Bases (VLDB), August 2013.

[82] Alex Dean: "Improving Snowplow's Understanding of Time," snowplowanalytics.com, September 15, 2015.

[83] "Windowing (Azure Stream Analytics)," Microsoft Azure Reference, msdn.microsoft.com, April 2016.

[84] "State Management," Apache Samza 0.10 Documentation, samza.apache.org, December 2015.

[85] Rajagopal Ananthanarayanan, Venkatesh Basker, Sumit Das, et al.: "Photon: Fault-Tolerant and Scalable Joining of Continuous Data Streams," at ACM International Conference on Management of Data (SIGMOD), June 2013. doi:10.1145/2463676.2465272

[86] Martin Kleppmann: "Samza Newsfeed Demo," github.com, September 2014.

[87] Ben Kirwin: "Doing the Impossible: Exactly-Once Messaging Patterns in Kafka," ben.kirw.in, November 28, 2014.

[88] Pat Helland: "Data on the Outside Versus Data on the Inside," at 2nd Biennial Conference on Innovative Data Systems Research (CIDR), January 2005.

[89] Ralph Kimball and Margy Ross: The Data Warehouse Toolkit: The Definitive Guide to Dimensional Modeling, 3rd edition. John Wiley & Sons, 2013. ISBN: 978-1-118-53080-1

[90] Viktor Klang: "I'm coining the phrase 'effectively-once' for message processing with at-least-once + idempotent operations," twitter.com, October 20, 2016.

[91] Matei Zaharia, Tathagata Das, Haoyuan Li, et al.: "Discretized Streams: An Efficient and Fault-Tolerant Model for Stream Processing on Large Clusters," at 4th USENIX Conference in Hot Topics in Cloud Computing (HotCloud), June 2012.

[92] Kostas Tzoumas, Stephan Ewen, and Robert Metzger: "High-Throughput, Low-Latency, and Exactly-Once Stream Processing with Apache Flink," data-artisans.com, August 5, 2015.

[93] Paris Carbone, Gyula Fóra, Stephan Ewen, et al.: "Lightweight Asynchronous Snapshots for Distributed Dataflows," arXiv:1506.08603 [cs.DC], June 29, 2015.

[94] Ryan Betts and John Hugg: Fast Data: Smart and at Scale. Report, O'Reilly Media, October 2015.

[95] Flavio Junqueira: "Making Sense of Exactly-Once Semantics," at Strata+Hadoop World London, June 2016.

[96] Jason Gustafson, Flavio Junqueira, Apurva Mehta, Sriram Subramanian, and Guozhang Wang: "KIP-98 – Exactly Once Delivery and Transactional Messaging," cwiki.apache.org, November 2016.

[97] Pat Helland: "Idempotence Is Not a Medical Condition," Communications of the ACM, volume 55, number 5, page 56, May 2012. doi:10.1145/2160718.2160734

[98] Jay Kreps: "Re: Trying to Achieve Deterministic Behavior on Recovery/Rewind," email to samza-dev mailing list, September 9, 2014.

[99] E. N. (Mootaz) Elnozahy, Lorenzo Alvisi, Yi-Min Wang, and David B. Johnson: "A Survey of Rollback-Recovery Protocols in Message-Passing Systems," ACM Computing Surveys, volume 34, number 3, pages 375–408, September 2002. doi:10.1145/568522.568525

[100] Adam Warski: "Kafka Streams – How Does It Fit the Stream Processing Landscape?," softwaremill.com, June 1, 2016.