آموزش کافکا | کافکا ( Kafka) چیست؟ | چگونه کافکا را در آپاچی راه اندازی کنیم؟ چه مفاهیمی در راه اندازی Kafka وجود دارد؟ و ...
جریان‌پردازی ، تاریخچه‌اش بیشتر برمی‌گردد به سیستم‌ها یا فریمورکهایی که قابلیت پردازش یک یا چند دنباله از رخدادهای نامتناهی را فراهم می‌کنند. این سیستم‌ها برای پردازش این رخدادها، عموماً امکاناتی از قبیل جازدنِ (Plugin) منطق‌های شخصی‌شده (Customized Logic) فراهم می‌کردند، می‌توانید کارهایی از قبیل فیلتر کردن ، تجمیع‌سازی مبتنی بر پنجره‌های زمانی و یا الحاق کردن جریان‌ها را انجام دهید.
برخی فریمورکهایی هم وجود دارند که کارهایی مانند موازی‌سازی را برایتان انجام می‌دهند و دیگر نیاز نیست که خیلی نگران موازی‌سازی‌ها باشید. این فریم‌ورک‌ها اغلب امکانات تحمل‌پذیری خطا هم فراهم می‌کنند. زیرا فریم‌ورک میتواند بر روی یک ماشین دیگر دوباره یک پراسس برایتان اجرا کنند. آپاچی کافکا توسط LinkedIn تولید شد و در سال 2011 به بنیاد Open source آپاچی پیوست.Kafka با Scala و Java نوشته شده است.
آپاچی Kafka مبتنی بر publish-subscribe که سیستم پیام رسانی است که بر اساس سرعت، مقیاس پذیری طراحی شده است.در Big Data ما حجم زیادی از دیتا را استفاده می کنیم با توجه به حجم زیاد دیتاها دو چالش اصلی در پیش رو داریم.چالش نخست: چگونه حجم زیادی از دیتا را جمع آوری کنیم و چالش دوم: چگونه دیتاهای جمع آوری شده را آنالیز کنیم. برای غلبه بر این چالش ها باید یک سیستم پیام رسانی داشته باشیم.
Kafka بر مبنای برنامه های توزیع شده طراحی شده است.حال برنامه های توزیع شده می تواند در چندین سیستم در شبکه در یک زمان معین (به طور همزمان) با هماهنگ سازی بین خود برای تکمیل یک کار خاص، به شیوه ای سریع و کارآمد اجرا شود معمولاً، وظایف پیچیده و وقت گیر که با یک برنامه غیر توزیع شده(اجرا در یک سیستم واحد) به اتمام می رسد، می تواند در عرض چند ثانیه توسط یک برنامه توزیع شده با استفاده از قابلیت های محاسباتی انجام می شود. Kafka به عنوان یک جایگزین مناسب برای broker message های سنتی استفاده می شود در مقایسه با دیگر سیستم های پیام، Kafka دارای قابلیت های fault tolerant, replication, built-inpartitioning بهتری است که آن را مناسب برای پردازش پیام هایی با مقیاس بزرگ می کند.
یک سیستم پیام رسان وظیفه انتقال دیتا ازبرنامه ای به برنامه دیگر را دارد بنابراین این برنامه ها تمرکزشان را روی دیتا می گذارند ولی نکته اینجاست که چگونه این دیتاها را به اشتراک می گذارند؟!پیام های توزیع شده بر مبنای مفهومی به نام reliable message queuying یا صف های پیام های قابل اعتماد است پیام به صورت یکنواخت در صف هایی بین برنامه های کلانیت و سیستم پیام رسان قرار می گیرند.دو نوع الگوی پیام رسان در دسترس است point-to-point و دیگری publish-subscribe(pub-sub) سیستم پیام.
پیام ها در صف هایی قرار می گیرند و یک یا چند مصرف کننده می تواند از پیام ها استفاده کنند اما تنها یک پیام خاص را یک مصرف کننده می تواند مصرف کند هنگامی که یک مصرف کننده یک پیام را در صف می خواند پیام از آن صف می تواند خارج شود.مثال از این نمونه سیستم ها، order processingها هستند که به ترتیب هر پروسه به وسیله یک order processپردازش می شود اما چندین order process می تواند همزمان با هم کار کند نمودار زیر این ساختار را نشان می دهد.
در این سیستم پیام ها بر اساس topic یا موضوعات هستند، برخلاف point-to-point system زمانی که مصرف کنندگان از پیام ها استفاده می کنند پیام ها باقی می مانند پیام ها در یک یا چند topic هستند و مصرف کنندگان به همه ی پیام هایی که به topicهای مختلف است دسترسی دارند.
در publish-subscribe system تولیدکنندگان پیام ها را publish و مصرف کنندگان که پیام ها را دریافت می کنند را subscriber می گویند.برای مثال در زندگی ما از این متدودها در دیش ها استفاده می شود که کانال های مختلفی مانند ورزشی، فیلم ها و موسیقی را منتشر می کنند و هر کس می تواند به مجموعه ای از کانال های خود دسترسی داشته باشد و هر زمان که کانال های مشترکی از آن ها در دسترس باشد آن ها را دریافت کند.
Kafka واقعاً برای حجم زیاد داده طراحی شده است، سیستم‌های قدیمی تر عموماً تنها مسئول ذخیره‌سازی داده‌هایی بودند که در پایگاه داده تولید می‌شد اما Kafka برای ذخیره‌سازی مواردی از قبیل آمارهای سنجش کسب‌وکار (Business Metrics)، لاگ‌های سرویس‌ها، آمارهای سنجش عملیاتی (Operational Metrics) و … بوده است، این نوع داده‌ها از لحاظ حجم ۱۰۰ یا ۱۰۰۰ برابر بزرگ‌تر از داده‌هایی هستند که در پایگاه داده ذخیره می‌کنید.چ
این‌ها چیزهایی نیست که سیستم‌های پیام‌رسانی مانند Active MQ و RabbitMQ برایش طراحی شده باشد اما Kafka واقعاً برای این‌ها طراحی شده است. برای مثال Kafka از ابتدا به عنوان یک سیستم توزیع‌شده طراحی شده است بنابراین اگر حجم داده‌ها افزایش یابد می‌توانید به راحتی ماشین‌های بیشتری به کلاستر اضافه کنید تا آن حجم داده را رسیدگی کند.
لازم به ذکر است که Kafka بسیار سریع عمل می کند و از downtime و از دست دادن دیتاها جلوگیری می کند.Kafka را می توان در بسیاری از موارد استفاده کرد که برخی از آن ها به شرح زیر است:
اغلب برای مانیتور کردن دیتاهای عملیاتی استفاده می شود که شامل جمع آوری دیتا از برنامه های توزیع شده برای تولید منابع متمرکز از دیتاهای عملیاتی است.
می تواند logهای سرویس های مختلف را در سراسر سازمان جمع آوری کند و آن را به فرمت استاندارد در دسترسی قرار دهد.
frame workهایی از قبیل streaming, spark, storm دیتا را از یک موضوع یا topic می خواند و آن را پردازش می کند دیتای پردازش شده را در یک topic جدید می نویسند تا برای userها و applicationها در دسترس باشند.
پیش از اینکه به جزئیات Kafka وارد شویم باید درباره ی اصطلاحات اصلی از قبیل producers, broker, topic و consumers آگاهی داشته باشیم.دیاگرام زیر نشان می دهد اصطلاحات اصلی با جزئیات هر مؤلفه را:
در بالای دیاگرام topic قرار دارد که 3 پارتیشن در داخل آن قرار دارد پارتیشن 1 دو تا فاکتور offset، 0 و 1 دارد پارتیشن 2 چهار تا فاکتور offset، 0 و 1 و 2 و 3 دارد و پارتیشن 3 یک فاکتور offset، 0 دارد، id شناسه سرور است که به عنوان host قرار گرفته است.فرض کنید اگر فاکتور تکرار یا topic, replicate به 3 تنظیم شده باشد، Kafka 3 تکه یکسان از هر پارتیشن ایجاد می کند و آن ها را کلاستر قرار می دهد تا برای تمام عملیات operation در دسترس قرار گیرد.برای تعادل کردن load در هر کلاسترها یک یا چند پارتیشن را ذخیره می کنند. Producer و consumerها نیز می توان پیام ها را در همان زمان publish و بازیابی کنند.
stream پیام هایی که متعلق به یک دسته خاصی هستند topic نامیده می شوند و دیتاها در topicها ذخیره می شود.Topicها به پارتیشن هایی تقسیم می شوند برای هر topic Kafka حداقل یک پارتیشن را در نظر می گیرد که هر یک از این پارتیشن ها شامل پیام های هستند که دارای ترتیب یکسان و تغییرناپذیراند.لازم به ذکر است که یک پارتیشن به عنوان مجموعه ای از فایل های قطعه قطعه شده با اندازه های یکسان اجرا می شود.
Topic ها ممکن است چندین پارتیشن داشته باشند بنابراین می تواند مقدار دلخواه ای از داده ها را مدیریت کند.Partition offset هر پارتیشن پیام، یک شناسه منحصر به فردی دارد که offset نامیده می شود.
Backup جزئی از یک پارتیشن نیست و هرگز در پروسه خواندن و نوشتن دیتا دخالتی ندارد backup را برای جلوگیری از دست رفتن دیتاها استفاده می کنیم.
Broker سیستم های ساده ای هستند که مسئول نگهداری دیتاهای published شده می باشد هر broker ممکن است دارای یک یا چند پارتیشن در هر topic باشد.فرضیه 1 اگر Nپارتیشن در یک topicوجود داشته باشد،Nتعداد brokerو هر broker میتواند یک پارتیشن داشته باشد.فرضیه 2 اگر Nپارتیشن در یک topic وجود داشته باشد بیش از N broker یعنی (n +m) پس N broker یک پارتیشن دارد و broker M هیچ پارتیشنی برای topic خاصی ندارد.فرضیه 3 اگر N پارتیشن در یک topic وجود داشته باشد کمتر از broker N یعنی (N-M) پس هر broker یک یا چند پارتیشن دارد که در میان آن ها به اشتراک می گذارد لازم به ذکر است که این سناریو به دلیل توزیع بار به صورت نابرابر میان broker ها توصیه نمی شود.
Kafkaهایی را که دارای بیش از یک broker هستند را Kafka cluster نامیده می شوند.یک Kafka clusterرا می توان بدون هیچ down time گسترش داد. از کلاسترها برای مدیریت پایداری و replication of message دیتا استفاده کرد.Producerها پیامها را در یک یا چند Kafka topic ، Publish می کنند.Producerها دیتاها را به Kafka brokerها ارسال می کنند.زمانی که Producerیک پیام را Publish می کند به یک broker، broker پیام را به یک پارتیشن اضافه می کند. همچنین می تواند پیام ها را به یک پارتیشن که خود انتخاب می کند ارسال کند.
consumerها دیتاها را از brokerها می خوانند. Consumerها subscribeهایی هستند که یک یا چند topic، پیامی کهpublish شده است را مصرف می کنند و اینکار را به وسیله گرفتن دیتا از brokerها انجام می دهند.
Leader یکnode است که مسئول خواندن دیتاها و نوشتن دیتاها در پارتیشن ها است. هر پارتیشن دارای یک سرور است به عنوان Leader عمل می کند.
node ای است که دستورالعمل های Leader را دنبال می کند به عنوان follower خوانده می شود اگر یک Leader، fail شود یک follower به صورت خودکار به یک leader جدید تبدیل می شود.Follower خود نیز به عنوان یک مصرف کننده عمل می کند که پیام را می گیرد و data store خود را به روز رسانی می کند.
دیاگرام زیر Kafka Cluster را نشان می دهد.
Kafka Cluster معمولاً شامل چندین broker است که برای نگهداری load balance از آن استفاده می شود. Kafka broker ها stateless هستند بنابراین از zookeeper برای حفظ Clusterهای خود استفاده می کند. برای نمونه یک Kafka broker می تواند هزاران دیتا را در هر ثانیه بخواند و بنویسد.انتخاب Kafka broker leader توسط zookeeper انجام می شود.
Zookeeper برای مدیریت و هماهنگی Kafka brokerها استفاده می شود سرویس Zookeeper برای آگاهی از producerها و consumerها و همچنین وجود هر broker جدید در سیستم Kafka و یا fail شدن سیستم Kafka استفاده می شود.Notificationهای دریافت شده توسط zookeeperها در مورد موجود بودن یا fail شدن broker ها خبر می دهد به این ترتیب producerها و consumerها تصمیم می گیرند برای هماهنگ کردن کار خود از broker دیگری استفاده کنند.
Producerها دیتا ها را از brokerها می گیرند زمانی که broker جدید start می شود همه producerها به طور خودکار شروع به جست و جو broker جدید کرده و پیام ها را به آن ارسال می کنند نکته: Kafka producer منتظر تایید پیام ها توسط broker نمی ماند و خیلی سریع پیام ها را ارسال می کند و این broker است که پیام ها را مدیریت می کند.
از آن جایی که brokerها stateless هستند. Consumer باید به وسیله پارتیشن، تعداد پیام های که مصرف کرده را حفظ کند در صورتی که consumer پیام offset را تایید کند. نشان دهنده این است که consumerها همه ی پیام های قبلی را مصرف کرده اند در این حالت consumer یک درخواست غیر مستقیم به broker می دهد تا یک buffer byte آماده مصرف داشته باشد. consumer می تواند به هر نقطه ای از این پارتیشن به وسیله مقدار offset برود.
توجه داشته باشید که مقدار offset توسط zookeeper به consumer اطلاع داده می شود. تا اینجا درباره مفاهیم اصلی Kafka بحث کردیم حال می خواهیم در جریان workflow در Kafka قرار بگیریم.Kafka مجموعه ای از topicهاست که به یک یا چند پارتیشن تقسیم می شود به صورت توالی خطی از پیام ها است هر یک از پیام ها توسط شاخصی شناسایی می شوند که offset نامیده می شوند.تمام دیتاها در Kafka cluster در پارتیشن های مجزا هستند پیام های ورودی در انتهای پارتیشن نوشته می شود و پیام ها توسط consumerها خوانده می شود.
مراحل Work flow of pub-sub messaging به شرح زیر است:
در یک سیم پیام رسان به جای یک consumerها گروهی از consumerها با یک group ID برای یک topic مشترک را خواهیم داشت. به عبارت دیگر consumerها در یک topic با همان group ID مشترک به عنوان یک گروه در نظر گرفته می شوند و پیام ها در میان آن ها به اشتراک گذاشته می شوند.مراحل آن به شرح زیر است:
یکی از وابستگی های میانی apache zookeeper, apache Kafka است که یک سرویس هماهنگ ساز توزیع شده است. Zookeeper سرویسی به عنوان رابط هماهنگ کننده بین Kafka brokerها consumerها است.Meta data, Kafkaها را در zookeeper ذخیره می کند، مانند اطلاعاتی در مورد topicها، brokerها و consumerها و offsetو ...Zookeeper یک سرویس هماهنگ کننده توزیع شده برای مدیریت مجموعه زیادی از hostها نیز می باشد.
هماهنگی و مدیریت یک سرویس در یک محیط توزیع شده یک فرآیند پیچیده است که zookeeper این موضوع را با معماری ساده خود حل کرده است.لازم به ذکر است zookeeper به توسعه دهندگان این اجازه را می دهد تا بر روی منطق برنامه های کاربردی متمرکز شوند بدون اینکه نگران ماهیت توزیع شده برنامه باشند.
پیش از نصب Kafka باید java بر روی ماشین شما نصب شده باشد اگر java بر روی ماشین شما نصب بوده باشد می توانیم ورژن این برنامه را با دستور زیر مشاهده کنیم.
Java -version#
اگر جاوا بر روی سیستم نصب نیست از لینک زیر JDK را دانلود کرده با توجه به مراحل زیر نصب را انجام دهید.
http://www.oracle.com/technet/work/java/javase/downloads/index.html
حال فایل JDK که دانلود کردیم را با دستور زیر extract می کنیم.
tar -xvf jdk-su201-linux-x64.tar.gz#
در این مرحله با توجه به محل قرارگیری برنامه jDK دستورات زیر را وارد کرده تا برنامه برای ما نصب شود.
alternatives --install usrbinjava java optjd. k1.8.0-201bin/java2 #
alternatives—config java#
اکثر برنامه های مبتنی بر جاوا از متغیرهای محیطی برای کارهای خود استفاده می کنند متغیر محیطی java را با استفاده از دستورات زیر وارد می کنیم.
پس از نصب جاوا نوبت به نصب zookeeper می رسد که با توجه به مراحل زیر نصب zookeeper را انجام می دهیم.
برای نصب zookeeper می توان آن از لینک زیر دانلود کرد.
http://zookeeper.apache.org/releases.html
فایل zookeeper را با دستور زیر extract می کنیم.
یک دایرکتوری به نام data برای data dir که در zookeeper config file مورد نیاز است را داخل دایرکتوری zookeeper می سازیم.
در این مرحله configuration file مربوط به zookeeper را تنظیم می کنیم ابتدا وارد دایرکتوری conf شده و فایل zoo-sample.ofg را به zoo.ofg تغییر نام می دهیم.
فایل zoo.ofg را با ویرایشگر vi باز کرده و موارد زیر را در این فایل درج نمایید.
حال برنامه zookeeper را start می کنیم.
در این مرحله بعد از start کردن zookeeper با دستور زیر به zookeeper، وصل می شویم.
پس از نصب zookeeper حال نوبت start کردن Kafka است.
برای start کردن Kafka دستور زیر را وارد نمایید.
در این مرحله ما یک Topic به نام test با یک پارتیشن و یک replication می سازیم برای اینکار باید:
ما با دستور زیر Topic جدیدی را که ساختیم را در لیست Topicهایمان ببینیم.
Kafka با خط فرمانی که دارد می توان یک فایل یا ورودی استاندارد را در آن وارد کرد. این ورودی به Kafka cluster ارسال می شود.
برای انجام ارسال پیام ،دستور و پیام زیر را وارد کنید.
اگر هر یک از دستورات فوق را در یک ترمینال دیگر اجرا کنید می توانید پیام ها را در ترمینال تولید کننده چاپ و آن ها را در ترمینال مصرف کننده مشاهده کنید.
تا این لحظه ها بر روی یک broker کار کردیم اما حال می خواهیم clusterهایمان به 3 node را گسترش دهیم. پس باید
حال فایل های جدید را با ویرایشگر vi باز کرده و تنظیمات را اعمال می کنیم.
id, broker id منحصر به فرد و دائمی مربوط به هر cluster است.ما part و دایرکتوری log را تغییر دادیم تا brokerها تلاش کنند از همان دستگاه اصلی register,port کنند یا داده های یکدیگر را بازنویسی کنند.
حالا یک topic جدید با Replication factor ایجاد می کنیم.
برای اینکه بدانیم در یک broker, cluster چه کاری انجام می دهد دستور زیر را وارد نمایید.
حال دستور قبل را برای test, topic که ایجاد کردیم اجرا می کنیم.
حال برای انجام تست چند پیام را برای topic جدید ارسال می کنیم
این پیام را می بینید
برای تست broker, fault tolerance که به عنوان leader عمل می کند آن را kill می کنیم
Leader تبدیل به یک slave شده و replication آن با sync node 1 نیست.
اما پیام ها هنوز هم برای consumerها در دسترس هستند حتی اگر leader در ابتدا پیام را نوشته و بعد down شده باشد برای مشاهده این موضوع دستور زیر را اجرا کنید.
![علیرضا قنبری پور](data:image/svg+xml;utf8,<svg xmlns="http://www.w3.org/2000/svg" width="120" height="120"></svg>)