فصل ۳ — طراحی یک پروژهٔ Kafka

آنچه در این فصل می‌آید

• طراحی یک پروژهٔ Kafka دنیای واقعی
• تعیین قالب داده
• مسائل موجود که بر استفاده از داده تأثیر می‌گذارند
• تصمیم‌گیری دربارهٔ زمان transformation داده
• کمک Kafka Connect برای شروع مسیر data streaming

در فصل قبل دیدیم چگونه از command line با Kafka کار کنیم و client Java استفاده کنیم. اکنون آن مفاهیم اولیه را گسترش می‌دهیم و راه‌حل‌های مختلف با Kafka را طراحی می‌کنیم. سؤالاتی را که هنگام layout استراتژی پروژهٔ نمونه‌ای که در این فصل شروع می‌کنیم باید در نظر بگیریم بحث می‌کنیم. هنگام توسعهٔ راه‌حل‌ها به خاطر داشته باشید، مثل بیشتر پروژه‌ها، ممکن است تغییرات جزئی در مسیر داشته باشیم و فقط به دنبال نقطه‌ای برای jump in و شروع توسعه هستیم. پس از این فصل برای حل use caseهای دنیای واقعی و تولید designی که exploration بیشتر Kafka در بقیهٔ کتاب را تسهیل کند آماده‌اید. مسیر یادگیری هیجان‌انگیز را شروع کنیم!

۳.۱ طراحی یک پروژهٔ Kafka

شرکت‌ها و پروژه‌های جدید می‌توانند از ابتدا Kafka را به کار ببرند، اما برای همهٔ adopters Kafka اینطور نیست. برای کسانی که در محیط‌های enterprise یا با سیستم‌های legacy کار کرده‌اند (و هر چیز بالای پنج سال احتمالاً legacy محسوب می‌شود)، شروع از صفر luxury نیست که همیشه داریم. با این حال، یکی از مزایای معماری موجود لیست pain pointهایی است که می‌توانیم به آن‌ها بپردازیم. contrast هم به shift در تفکر دربارهٔ داده در کارمان کمک می‌کند. در این فصل روی پروژه‌ای برای شرکتی کار می‌کنیم که آماده shift از روش فعلی مدیریت داده و اعمال این hammer جدید به نام Kafka است.

۳.۱.۱ تحویل گرفتن معماری دادهٔ موجود

برای background مثال fictional الهام‌گرفته از usage رو به رشد Kafka، topic Confluent در فصل ۱ (https://www.confluent.io/use-case/internet-of-things-iot/) و مقالهٔ عالی Janakiram MSV با عنوان «Apache Kafka: The Cornerstone of an Internet-of-Things Data Platform» استفاده sensor را شامل می‌شود [۱]. با topic sensor به‌عنوان use case، پروژهٔ fictional را بررسی می‌کنیم.

شرکت consulting fictional جدیدمان تازه contract کمک re-architect کارخانه‌ای را برده که روی e-bike کار می‌کند و آن‌ها را remotely مدیریت می‌کند. sensor در سراسر دوچرخه قرار دارند و continuously event دربارهٔ وضعیت تجهیزات داخلی که monitor می‌کنند می‌دهند. با این حال event آن‌قدر زیاد تولید می‌شود که سیستم فعلی بیشتر پیام‌ها را نادیده می‌گیرد. از ما خواسته شده potential داده را برای برنامه‌های مختلف unlock کنیم. علاوه بر این، دادهٔ فعلی شامل سیستم‌های relational database سنتی بزرگ و clustered است. با این همه sensor و پایگاه دادهٔ موجود، چگونه معماری Kafka-based جدید را بدون impact روی manufacturing بسازیم؟

۳.۱.۲ اولین تغییر

احتمالاً بهترین راه شروع task با big-bang نیست — همهٔ داده لازم نیست یک‌جا به Kafka برود. اگر امروز پایگاه داده دارید و فردا streaming data را test می‌کنید، یکی از آسان‌ترین on-rampها Kafka Connect است. اگرچه load production را handle می‌کند، لازم نیست از همان ابتدا. یک جدول پایگاه داده را می‌گیریم و معماری جدید را شروع می‌کنیم در حالی که برنامه‌های موجود فعلاً اجرا می‌مانند. اما اول چند مثال برای آشنایی با Kafka Connect.

۳.۱.۳ ویژگی‌های built-in

هدف Kafka Connect کمک به جابه‌جایی داده به یا از Kafka بدون نوشتن producer و consumer خودمان است. Connect frameworkی است که بخشی از Kafka است و استفاده از قطعات از پیش ساخته برای شروع streaming را ساده می‌کند. این قطعات connector نامیده می‌شوند و برای کار reliable با منابع دادهٔ دیگر توسعه یافته‌اند [۲].

اگر از فصل ۲ به یاد دارید، کد Java client producer و consumer که به‌عنوان مثال استفاده کردیم نشان می‌دهد Connect چگونه آن مفاهیم را با استفادهٔ داخلی از آن‌ها abstract می‌کند. یکی از آسان‌ترین شروع‌ها دیدن نحوهٔ بردن فایل log معمولی برنامه به topic Kafka است. آسان‌ترین گزینه برای اجرا و test Connect روی ماشین local، standalone mode است. scaling بعداً اگر standalone را دوست داشتیم!

در folder نصب Kafka، فایل‌های زیر تحت directory config را پیدا کنید:

• connect-standalone.properties
• connect-file-source.properties

در connect-standalone.properties key و value config آشنا از propertyهای client Java فصل ۲ را می‌بینید، از جمله bootstrap.servers. دانستن clientهای زیرین producer و consumer کمک می‌کند بفهمید Connect همان config را برای کارش استفاده می‌کند.

در مثال، داده را از یک data source می‌گیریم و در Kafka می‌گذاریم تا داده را از فایل Kafka-sourced بدانیم. با connect-file-source.properties included با نصب Kafka به‌عنوان template، فایلی به نام alert-source.properties می‌سازیم و متن فهرست ۳.۱ را داخل می‌گذاریم. این فایل config لازم برای setup فایل alert.txt و ارسال داده به topic مشخص kinaction_alert_connect را تعریف می‌کند. این مثال مراحل مشابه Connect Quickstart guide در https://docs.confluent.io/3.1.2/connect/quickstart.html را دنبال می‌کند. برای جزئیات بیشتر presentation Randall Hauch (Apache Kafka committer و PMC) از Kafka Summit (San Francisco, 2018) در http://mng.bz/8WeD توصیه می‌شود.

با config (نه code) می‌توانیم داده را از هر فایلی به Kafka ببریم. چون خواندن از فایل task رایج است، از classهای prebuilt Connect استفاده می‌کنیم — اینجا FileStreamSource [۲]. فرض کنید برنامه‌ای alert را به فایل متنی می‌فرستد.

فهرست ۳.۱ — پیکربندی Connect برای file source

name=alert-source
connector.class=FileStreamSource
tasks.max=1
file=alert.txt
topic=kinaction_alert_connect

مقدار property topic مهم است — بعداً برای verify کردن که پیام‌ها از فایل به topic kinaction_alert_connect کشیده شده‌اند استفاده می‌کنیم. فایل alert.txt برای تغییرات monitor می‌شود. tasks.max=1 چون فقط یک task برای connector لازم است و در این مثال parallelism مهم نیست.

توجه اگر ZooKeeper و Kafka را locally اجرا می‌کنید، مطمئن شوید brokerهای Kafka هنوز running هستند (اگر بعد از فصل قبل shutdown کردید).

اکنون config آماده است؛ Connect را start می‌کنیم و config را می‌فرستیم. با invoke کردن connect-standalone.sh و پارامتر فایل config سفارشی [۲]:

فهرست ۳.۲ — راه‌اندازی Connect برای file source

bin/connect-standalone.sh config/connect-standalone.properties \
  alert-source.properties

در terminal دیگر فایل متنی alert.txt در directoryی که Connect را start کردیم بسازید و چند خط با editor اضافه کنید. با دستور console-consumer از topic kinaction_alert_connect consume کنید [۲]:

فهرست ۳.۳ — تأیید رسیدن پیام‌های فایل به Kafka

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9094 \
  --topic kinaction_alert_connect --from-beginning

Connect باید محتوای فایل را ingest و به Kafka produce کند.

قبل از connector دیگر، sink connector و نحوهٔ برگرداندن پیام Kafka به فایل دیگر را بگوییم. چون مقصد (sink) فایل دیگر است، connect-file-sink.properties را ببینید. تغییر کوچک در فهرست ۳.۴ — outcome جدید به فایل نوشته می‌شود نه خوانده. FileStreamSink نقش sink را تعریف می‌کند. topic kinaction_alert_connect منبع داده است. متن فهرست ۳.۴ را در alert-sink.properties بگذارید [۲].

فهرست ۳.۴ — پیکربندی Connect برای file source و sink

name=alert-sink
connector.class=FileStreamSink
tasks.max=1
file=alert-sink.txt
topics=kinaction_alert_connect

اگر Connect هنوز running است، terminal را ببندید یا Ctrl-C بزنید. با file-source و file-sink propertyها restart کنید [۲]:

فهرست ۳.۵ — راه‌اندازی Connect برای file source و sink

bin/connect-standalone.sh config/connect-standalone.properties \
  alert-source.properties alert-sink.properties

برای تأیید sink، فایل alert-sink.txt را باز کنید و پیام‌های source را که به topic Kafka فرستاده شده ببینید.

۳.۱.۴ داده برای invoiceها

requirement دیگر: invoice سفارش دوچرخه. Connect به کسانی که connector سفارشی می‌سازند اجازه share با دیگران می‌دهد. پس از connector (فهرست‌های ۳.۴ و ۳.۵)، integrate کردن connector دیگر با standardize شدن تعامل Connect ساده‌تر است.

برای مثال manufacturing، source connector موجود که به‌روزرسانی جدول از پایگاه داده local را به topic Kafka stream می‌کند. goal تغییر کل معماری پردازش داده یک‌جا نیست — نشان می‌دهیم چگونه update از برنامهٔ table-based database را بیاوریم و برنامهٔ جدید parallel توسعه دهیم در حالی که سیستم دیگر as is می‌ماند. این مثال مراحل مشابه guide در https://docs.confluent.io/kafka-connect-jdbc/current/source-connector/index.html دارد.

اول پایگاه داده برای مثال‌های local setup می‌کنیم. برای سهولت از connector Confluent برای SQLite استفاده می‌کنیم. اگر sqlite3 در terminal prompt می‌دهد آماده‌اید؛ وگرنه SQLite را نصب کنید.

نکته فایل Commands.md در source code این فصل دستور نصب Confluent CLI و JDBC connector با confluent-hub را دارد. بقیهٔ مثال‌ها به directory نصب Confluent اشاره می‌کنند، نه directory نصب Kafka.

با sqlite3 kafkatest.db پایگاه داده می‌سازیم. سپس فهرست ۳.۶ برای جدول invoices و test data:

فهرست ۳.۶ — ساخت جدول invoices

CREATE TABLE invoices(
   id INT PRIMARY KEY     NOT NULL,
   title          TEXT    NOT NULL,
   details        CHAR(50),
   billedamt      REAL,
   modified     TIMESTAMP DEFAULT (STRFTIME('%s', 'now')) NOT NULL
);

INSERT INTO invoices (id,title,details,billedamt)
  VALUES (1, 'book', 'Franz Kafka', 500.00 );

هنگام طراحی جدول به capture تغییرات به Kafka فکر کنید. بیشتر use caseها کل database را نمی‌خواهند، فقط تغییرات بعد از initial load. timestamp، sequence number یا ID کمک می‌کند Connect به Kafka بگوید کدام داده modified شده [۳]. در فهرست ۳.۶ ستون‌های ID یا modified راهنما هستند.

فایل در etc/kafka-connect-jdbc/kafkatest-sqlite.properties — insert و update باعث پیام به Kafka می‌شود. source code فصل ۳ در Git repository دستور setup JDBC connector را دارد. JDBC connector بخشی از distribution Apache Kafka نیست مثل file connector. اگر format timestamp modified خطا داد، گزینه‌های دیگر در source code را ببینید.

Connect را با kafkatest-sqlite.properties start کنید:

فهرست ۳.۷ — راه‌اندازی Connect برای database table source

confluent-hub install confluentinc/kafka-connect-jdbc:10.2.0
confluent local services connect start
...
# See Commands.md for other steps
confluent local services connect connector config jdbc-source \
--config etc/kafka-connect-jdbc/kafkatest-sqlite.properties

فهرست ۳.۷ Connect را با Confluent CLI launch می‌کند. connect-standalone.sh هم قابل استفاده بود [۳]. قدرت Kafka Connect برای انتقال جدول database موجود به Kafka عالی است، اما sensorها (بدون پشتیبان database) technique دیگری می‌خواهند.

۳.۲ طراحی event sensor

connector موجود برای sensorهای state-of-the-art ما نیست؛ مستقیماً با event system آن‌ها از طریق producer سفارشی تعامل می‌کنیم. hook کردن و نوشتن producer برای ارسال داده به Kafka جایی است که requirementهای بخش‌های بعد را می‌بینیم.

شکل ۳-۱. setup کارخانه

هر gear یک major step در process — sensor به هر step — critical path — admin console دستور به sensor می‌دهد — quality check sensor قابل skip — فعلاً sensor event را به clustered database می‌فرستند؛ این بخش اصلی design است که می‌خواهیم تغییر دهیم

۳.۲.۱ مسائل موجود

دو چالش: data silo و recoverability.

مدیریت data silo

در کارخانه، داده و پردازش متعلق به یک برنامه است. دیگران برای استفاده باید با owner برنامه صحبت کنند. احتمال format قابل پردازش آسان چقدر است؟ یا اصلاً داده نمی‌دهد؟

shift از «data thinking» سنتی داده را برای همه در raw source در دسترس می‌گذارد. با دسترسی به داده هنگام ورود، نگران API برنامه که format خاص یا بعد از transformation سفارشی expose می‌کند نیستید. اگر برنامه API دادهٔ اصلی را اشتباه parse کند، untangle کردن ممکن است زمان‌بر باشد.

Recoverability

مزیت distributed system مثل Kafka این است که failure شرط expected است — planned و handled! علاوه بر blip سیستم، عنصر انسانی در توسعه هم هست. اگر برنامه defect یا logic issue داده را destroy کند، مسیر اصلاح چیست؟ با Kafka می‌تواند به‌ساده consume از ابتدای topic مثل --from-beginning در فصل ۲ باشد. data retention داده را بارها در دسترس نگه می‌دارد. reprocess برای correction قدرتمند است. اما اگر event اصلی در دسترس نباشد، retrofit سخت است.

event فقط یک‌بار از منبع sensor برای instance مشخص produce می‌شود. message broker نقش crucial در consumption pattern دارد. اگر در queuing system پیام بعد از خواندن subscriber از broker حذف شود (مثل application نسخه ۱.۰ در شکل ۳-۲)، از سیستم رفته. اگر defect logic بعداً پیدا شود، باید ببینید آیا با باقی‌ماندهٔ پردازش event اصلی قابل correction است — دوباره fire نمی‌شود. خوشبختانه broker Kafka گزینهٔ دیگری می‌دهد.

شکل ۳-۲. نگاه به اشتباه coding توسعه‌دهنده

Version 1.0: consumption پیام را حذف می‌کند — Version 1.1: Kafka همهٔ پیام‌ها را replay می‌کند چون چیزی حذف نشده

از نسخه ۱.۱، application می‌تواند پیام‌های already consumed را با logic جدید replay کند. کد اصلاح‌شده mistake نسخه ۱.۰ را دوباره پردازش می‌کند. replay هم نشان می‌دهد value چگونه در زمان تغییر کرده — parallel با write-ahead log (WAL): modifications قبل از apply در log نوشته می‌شوند؛ WAL در database systems رایج است.

چه زمانی داده تغییر کند؟

ترجیح: داده را اول به Kafka ببریم — خالص‌ترین form. هر step قبل از Kafka فرصت alter یا error formatting/logic است. hardware، software و logic در distributed computing fail می‌کنند — گرفتن داده اول در Kafka ability replay در failure را می‌دهد.

۳.۲.۲ چرا Kafka مناسب است

آیا Kafka در use case sensor fictional معنا دارد؟ این کتاب دربارهٔ Kafka است! اما چند دلیل compelling:

clientها می‌گویند database فعلی برای vertical scale گران شده. vertical scale یعنی افزایش CPU، RAM و disk روی ماشین موجود؛ horizontal scale یعنی سرور بیشتر. با horizontal scale cluster امید benefit بیشتر per buck. سرور broker شاید ارزان‌ترین نباشد، اما ۳۲ یا ۶۴ GB RAM روی آن‌ها load production را handle می‌کند [۴].

eventها continuously produce می‌شوند — شبیه تعریف stream processing. feed constant endpoint یا زمان توقف تعریف‌شده ندارد. پیام‌ها معمولاً زیر ۱۰ KB — پیام کوچک‌تر و page cache بیشتر performance سالم‌تر.

برخی developer امنیت‌محور ممکن است disk encryption built-in برای broker (data at rest) نباشد را دیده باشند — requirement فعلی سیستم نیست. اول سیستم را up می‌کنیم، security بعداً.

۳.۲.۳ نکات اولیه برای design

کدام feature برای نسخهٔ Kafka مشخص در دسترس است مهم است. نسخه 2.7.1 در مثال‌ها؛ برخی broker/client version را کنترل ندارند. جدول ۳.۱ milestoneهای major گذشته را highlight می‌کند [۵] — inclusive همهٔ versionها نیست.

جدول ۳.۱ — milestone نسخه‌های Kafka

Kafka version	Feature
2.0.0	ACL با prefix support و hostname verification (default برای SSL)
1.0.0	پشتیبانی Java 9 و بهبود JBOD disk failure
0.11.0.0	Admin API
0.10.2.0	بهبود client compatibility
0.10.1.0	جستجوی time-based
0.10.0.0	Kafka Streams، timestamp و rack awareness
0.9.0.0	امنیت (ACL، SSL)، Kafka Connect و consumer client جدید

broker از 0.10.0 با client version جدیدتر کار می‌کند — client را اول upgrade کنید، broker بعداً.

سؤالات برای فکر کردن به نحوهٔ process داده — تمرکز ساختار داده؛ implementation در فصل‌های بعد:

• آیا از دست دادن پیام acceptable است؟ (پرداخت mortgage vs RSS feed)
• آیا داده باید group شود؟ (تغییرات account با customer)
• آیا ترتیب تحویل essential است؟ (لغو سفارش قبل از سفارش)
• فقط last value یا history مهم است؟
• چند consumer — مستقل یا با ترتیب؟

۳.۲.۴ requirement داده کاربر

معماری جدید باید:

• capture پیام حتی اگر consuming service down باشد
• وقتی application از maintenance برمی‌گردد یا بعد از failure، دادهٔ لازم را داشته باشد
• status sensor به‌عنوان working یا broken (alert)
• ببینیم آیا بخشی از process bike به total failure منجر می‌شود
• history alert status sensor برای trend و پیش‌بینی failure
• audit log کاربرانی که update/query مستقیم به sensor می‌زنند
• برای compliance: چه کسی چه admin action روی sensor انجام داده

۳.۲.۵ plan سطح بالا برای اعمال سؤالات

برای audit log: همهٔ ورودی management API capture شود؛ فقط کاربران مجاز action کنند؛ پیام از دست نرود؛ grouping key لازم نیست — هر event مستقل. ترتیب در audit topic مهم نیست — timestamp در payload. Kafka sort by time دارد اما این use case warrant نمی‌کند.

شکل ۳-۳. use case audit

Console admin — Process step 1 sensor — audit event برای step 1 و 3 — Kafka brokers — software update event برای sensor step 1 و 3

جدول ۳.۲ — checklist audit

Kafka feature	Concern?
Message loss	Yes
Grouping	No
Ordering	No
Last value only	No
Independent consumer	Yes

alert trend: group با key — part ID هر stage. alert هر ۵ ثانیه — از دست دادن گاه‌گاه acceptable. «Needs Maintenance» هر چند روز برای trend failure.

شکل ۳-۴. use case alert trend

Sensor trend events by stage به Kafka — Hadoop HDFS، Kafka brokers — permanent storage برای analysis

جدول ۳.۳ — checklist alert trend

Kafka feature	Concern?
Message loss	No
Grouping	Yes
Ordering	No
Last value only	No
Independent consumer	Yes

alerting status: group با key (process stage)؛ فقط current status — نه history. «replace» دقیق نیست: Kafka event جدید را به انتهای log append می‌کند (immutable). ظاهر update با log compaction — فصل ۷. consumer usage: critical alert partition اختصاصی — uptime سریع.

شکل ۳-۵. use case alert

Sensor روی critical path failure — Kafka brokers — dashboard operator برای current status

جدول ۳.۴ — checklist alert

Kafka feature	Concern?
Message loss	No
Grouping	Yes
Ordering	No
Last value only	Yes
Independent consumer	No

plan requirement هم application را clarify می‌کند هم استفاده از Kafka را validate می‌کند.

۳.۲.۶ مرور blueprint

گروه‌های داده:

• Audit data
• Alert trend data
• Alert data

alert trend data ممکن است starting point برای alerts topic باشد — یک topic می‌تواند topic دیگر را populate کند. برای design اول: هر event type به topic منطقی خود — audit → audit topic، alert trend → alert trend topic، alert → alert topic. one-to-one mapping برای سادگی.

۳.۳ قالب داده

یکی از آسان‌ترین چیزهایی که skip می‌شود اما critical است: format داده. XML و JSON format استاندارد برای structure. اما حتی با syntax واضح، اطلاعات در داده missing است — معنای ستون اول؟ type فیلد دوم؟ schema این اطلاعات را برای code و برنامه‌های دیگر فراهم می‌کند.

در documentation Kafka به Apache Avro اشاره شده. Avro schema definition و storage در فایل Avro [۶]. Avro احتمالاً در کد Kafka دنیای واقعی می‌بینید — روی این choice تمرکز می‌کنیم.

۳.۳.۱ plan برای داده

مزیت Kafka: producer و consumer مستقیماً به هم tie نیستند. Kafka به‌صورت پیش‌فرض data validation نمی‌کند. اما هر process/application باید بداند داده چه معنایی و چه format دارد. schema به developer ساختار و intent را می‌دهد — نه README یا reverse-engineer از dump.

فهرست ۳.۸ — مثال schema Avro

{
      "type" : "record",
      "name" : "kinaction_libraryCheckout",
      ...
      "fields" : [{"name" : "materialName",
                   "type" : "string",
                   "default" : ""},
                  {"name" : "daysOverDue",
                   "type" : "int",
                   "default" : 0},
                  {"name" : "checkoutDate",
                   "type" : "int",
                   "logicalType": "date",
                   "default" : "-1"},
                  {"name" : "borrower",
                   "type" : {
                         "type" : "record",
                         "name" : "borrowerDetails",
                         "fields" : [
                            {"name" : "cardNumber",
                              "type" : "string",
                              "default" : "NONE"}
                          ]},
                          "default" : {}
                  }
      ]
}

field daysOverDue: int با default 0 — contract واضح برای producer و consumer. cardNumber string است نه number.

schema با Avro evolution داده را handle می‌کند — مثل Liquibase در relational database. داده probable تغییر می‌کند. schema از ابتدا لازم است؟ با scale بزرگ‌تر، correctness داده؟ consumer بیشتر → burden testing. شاید همه consumerها را نشناسیم.

۳.۳.۲ setup dependency

چرا Avro؟ Avro همیشه با schema serialize می‌شود [۷]. schema هنگام read/write؛ rule برای schema در حال تغییر. اگر JSON دیده‌اید، Avro آسان است — schema language هم JSON. schema تغییر کند، هنوز process می‌کنید — دادهٔ قدیمی schema قدیمی، جدید schema جدید. client سود می‌برند.

محبوبیت در Hadoop و جاهای دیگر. Confluent support [۶]. binding برای زبان‌های زیاد. بدون code generation هم dynamic Avro.

فهرست ۳.۹ — افزودن Avro به pom.xml

<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro</artifactId>
  <version>${avro.version}</version>
</dependency>

plugin برای generate Java source از schema:

فهرست ۳.۱۰ — افزودن Avro Maven plugin به pom.xml

<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>${avro.version}</version>
  <executions>
    <execution>
      <phase>generate-sources</phase>
      <goals>
        <goal>schema</goal>
      </goals>
      ...
    </execution>
  </executions>
</plugin>

schema alert status: فایل kinaction_alert.avsc — class Java Alert:

فهرست ۳.۱۱ — schema Alert: kinaction_alert.avsc

{
    ...
    "type": "record",
    "name": "Alert",
    "fields": [
      {
         "name": "sensor_id",
         "type": "long",
         "doc": "The unique id that identifies the sensor"
      },
      {
         "name": "time",
         "type": "long",
         "doc":
           "Time alert generated as UTC milliseconds from epoch"
      },
      {
         "name": "status",
         "type": {
            "type": "enum",
            "name": "AlertStatus",
            "symbols": [
              "Critical",
              "Major",
              "Minor",
              "Warning"
            ]
         },
         "doc":
          "Allowed values sensors use for current status"
     }
 ]
}

doc الزامی نیست اما برای developer آینده valuable است — مثلاً time: string؟ timezone؟ leap seconds؟ namespace → Java package. mvn generate-sources یا mvn install → Alert.java و AlertStatus.java.

برای producer/consumer: serializer Avro — مثال Confluent. dependency kafka-avro-serializer:

فهرست ۳.۱۲ — افزودن Kafka serializer به pom.xml

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>${confluent.version}</version>
</dependency>

repository Confluent در pom:

<repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
</repository>

HelloWorldProducer را برای Avro modify می‌کنیم — KafkaAvroSerializer برای value.serializer، topic kinaction_schematest، Schema Registry (فصل ۱۱):

فهرست ۳.۱۳ — producer با Avro serialization

public class HelloWorldProducer {

  static final Logger log =
    LoggerFactory.getLogger(HelloWorldProducer.class);

  public static void main(String[] args) {
    Properties kaProperties = new Properties();
    kaProperties.put("bootstrap.servers",
      "localhost:9092,localhost:9093,localhost:9094");
    kaProperties.put("key.serializer",
      "org.apache.kafka.common.serialization.LongSerializer");
    kaProperties.put("value.serializer",
      "io.confluent.kafka.serializers.KafkaAvroSerializer");
    kaProperties.put("schema.registry.url",
      "http://localhost:8081");

    try (Producer<Long, Alert> producer =
      new KafkaProducer<>(kaProperties)) {
      Alert alert =
        new Alert(12345L,
          Instant.now().toEpochMilli(),
          Critical);

      log.info("kinaction_info Alert -> {}", alert);

      ProducerRecord<Long, Alert> producerRecord =
          new ProducerRecord<>("kinaction_schematest",
                               alert.getSensorId(),
                               alert);

      producer.send(producerRecord);
    }
  }
}

تفاوت‌ها minor: type Producer و ProducerRecord، config value.serializer. consumption: KafkaAvroDeserializer [۱۰]:

فهرست ۳.۱۴ — consumer با Avro serialization

public class HelloWorldConsumer {

    final static Logger log =
      LoggerFactory.getLogger(HelloWorldConsumer.class);

    private volatile boolean keepConsuming = true;

    public static void main(String[] args) {
      Properties kaProperties = new Properties();
      kaProperties.put("bootstrap.servers", "localhost:9094");
      ...
      kaProperties.put("key.deserializer",
        "org.apache.kafka.common.serialization.LongDeserializer");
      kaProperties.put("value.deserializer",
        "io.confluent.kafka.serializers.KafkaAvroDeserializer");
      kaProperties.put("schema.registry.url", "http://localhost:8081");

        HelloWorldConsumer helloWorldConsumer = new HelloWorldConsumer();
        helloWorldConsumer.consume(kaProperties);

        Runtime.getRuntime()
          .addShutdownHook(
             new Thread(helloWorldConsumer::shutdown)
          );
    }

    private void consume(Properties kaProperties) {
        try (KafkaConsumer<Long, Alert> consumer =
          new KafkaConsumer<>(kaProperties)) {
          consumer.subscribe(
              List.of("kinaction_schematest")
          );

          while (keepConsuming) {
            ConsumerRecords<Long, Alert> records =
              consumer.poll(Duration.ofMillis(250));
            for (ConsumerRecord<Long, Alert> record :
              records) {
              log.info("kinaction_info offset = {}, kinaction_value = {}",
                record.offset(),
                record.value());
            }
          }
      }

    private void shutdown() {
      keepConsuming = false;
    }
}

consumer با update deserializer config و Avro تغییر زیادی نمی‌خواهد. اکنون what (هدف و format داده) را داریم — how در فصل بعد. topic schema در فصل ۱۱؛ object type در فصل‌های ۴ و ۵. ارسال داده به Kafka straightforward است اما config-driven behaviorهای زیادی requirement خاص را satisfy می‌کنند.

خلاصه

• طراحی راه‌حل Kafka با درک داده شروع می‌شود: data loss، ordering، grouping.
• نیاز به group داده تعیین می‌کند key کنیم یا نه.
• schema definition هم code generate می‌کند هم تغییرات آیندهٔ داده را handle می‌کند؛ با client سفارشی Kafka استفاده می‌شود.
• Kafka Connect connectorهای موجود برای read/write از منابع مختلف دارد.

منابع

1. J. MSV. «Apache Kafka: The Cornerstone of an Internet-of-Things Data Platform» (February 15, 2017). https://thenewstack.io/apache-kafka-cornerstone-iot-data-platform/ (accessed August 10, 2017).
2. «Quickstart.» Confluent documentation (n.d.). https://docs.confluent.io/3.1.2/connect/quickstart.html (accessed November 22, 2019).
3. «JDBC Source Connector for Confluent Platform.» Confluent documentation (n.d.). https://docs.confluent.io/kafka-connect-jdbc/current/source-connector/index.html (accessed October 15, 2021).
4. «Running Kafka in Production: Memory.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka/deployment.html#memory (accessed June 16, 2021).
5. «Download.» Apache Software Foundation (n.d.). https://kafka.apache.org/downloads (accessed November 21, 2019).
6. J. Kreps. «Why Avro for Kafka Data?» Confluent blog (February 25, 2015). https://www.confluent.io/blog/avro-kafka-data/ (accessed November 23, 2017).
7. «Apache Avro 1.8.2 Documentation.» Apache Software Foundation (n.d.). https://avro.apache.org/docs/1.8.2/index.html (accessed November 19, 2019).
8. «Apache Avro 1.8.2 Getting Started (Java): Serializing and deserializing without code generation.» Apache Software Foundation (n.d.). https://avro.apache.org/docs/1.8.2/gettingstartedjava.html#download_install (accessed November 19, 2019).
9. «Apache Avro 1.8.2 Getting Started (Java): Serializing and deserializing without code generation.» Apache Software Foundation (n.d.). https://avro.apache.org/docs/1.8.2/gettingstartedjava.html#Serializing+and+deserializing+without+code+generation (accessed November 19, 2019).
10. «Application Development: Java.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/app-development/index.html#java (accessed November 20, 2019).
11. «Installation: Maven repository for jars.» Confluent documentation (n.d.). https://docs.confluent.io/3.1.2/installation.html#maven-repository-for-jars (accessed November 20, 2019).