نوشته های مریم محمدی

Apache-Kafka-بخش پنجم-(ساخت کلاستر)

مریم محمدی — Mon, 26 Aug 2024 15:22:33 +0330

ما تا الان نحوه کار کافکا، دمویی از اون و نصبش و جزییات بیشتری از کافکا رو با هم مشاهده کردیم. در این نوشتار میخوایم موارد پیشرفته ی دیگه ای مانند نحوه ساخت کلاستر کافکا و zookeeper رو با هم ببینیم. این جزییات برای ایجاد کلاستر در عملیات ضروری هستن. تعریفQuorumsحداقل تعداد نود هایی که در یک cluster فعال باشند و بتوانند با یکدیگر ارتباط داشته باشند، quorums نامیده میشود. از دو منظر میتوان این تعداد حداقل نودها را به دست آورد. اول از طریق اینکه حداقل تعداد نود در دسترس برای کلاستر که بتوانند با هم کار کنند. روش دیگر براساس یافتن حدلقل تعداد نود برای داشتن failure مورد نظر. با روش اول حداقل تعداد نود برابر است با n/2 +1 (نصف تعداد نودها به اضافه 1) بنابراین quorum در کلاستری با 5 نود برابر با 3 است. یعنی اگر این کلاستر به دو بخش بشکند با تعداد 3 نود مانند پیش میتواند کار کند. با روش دوم میگویند با از بین رفتن (n-1)/2 از نودها، همچنان کلاستر کار خود را انجام میدهد یعنی اگر کلاستر zookeeper ما 5 نود داشته باشد به از دست دادن 2 نود همچنان مقاوم است.نیازمندی های سخت افزاری و نرم افزاری Zookeeperاستفاده از Zookeeper در کافکا memory intensive نیست و در حد 8 گیگ مموری براش کافیه و با کمتر هم اوکیه. دیسک ولی براش خیلی اهمیت داره و برای اینکه latency کم باشد، از ssd ها استفاده بشود بهتر است.(برعکس کافکا که نحوه نوشتن و خواندنش sequential است و ssd بودن برایش فایده ای نخواهد داشت.)ارتباط با zookeeper از کافکا با نتورک هستش بنابراین network timeout برای بالا رفتن سرعت در اینجا اهمیت زیادی دارد.کلاستر Zookeeperبرای اینکه کلاستر راه بندازیم لازم داریم zookeeper هامون هم چند نود داشته باشند چون یک نود تنها در مقابل fault tolerance مقاوم نیست. من در اینجا 3 تا نود zookeeper بالا میارم. Zookeeper خودش cluster friendly نیست و لازم داره ما id های unique برای هر نود داخل کلاستر تعریف کنیم. اینکار با متغیر ZOO_MY_ID انجام میشه و port ای مخصوص هر نود تخصیص بدیم و تمامی سرورهای داخل کلاستر zookeeper را برای آن تعریف کنیم. فایل dicker compose برای ساخت این کلاستر zookeeper در ادامه آمده است. ما باید در فضای zoo_servers باید هر مقداری که به ZOO_MY_ID نسبت دادیم رفرنس بدیم بنابراین اعداد رو حتما لازمه دستی خودمون تعیین کنیم(بعضی image ها اجازه ی تولید رندم id رو دارند که برای کلاستر شدن مناسب نیست.) در فایل docker-compose زیر اعداد به صورت bold آمده است.میتونیم ZOO_SERVERS رو در فایل zoo.conf هم تعریف کنیم ولی وقتی به عنوان environment variable در اینجا قرار بدیم دم دستمونه و راحت میتونیم تغییرش بدیم و نیاز نیست حتما این فایل conf رو در هنگام عملیاتی شدن هم دست بزنیم. (در اینجا سرور اول همه مقادیرش دیفالته، برای بعدی یک واحد بیشتر کردم و سومی هم همینطور)باید حواسمون باشه حتمنی که وقتی داریم هر سروری رو کانفیگ میکنیم در بخش ZOO_SERVERS آدرس IP ش باید با 0.0.0.0 تنظیم بشه و کانفیگ های دیفالت بذاریم جون از داخل میخواد باهاش صحبت بشه (طوری آدرس میدیم که از داخل بتوان به آن وصل شد.) و بقیه با IP سیستمی که داکر روی اون بالا میاد(طوری آدرس میدیم که از بیرون بتوان به آن وصل شد.) در اینجا ip سیستمم 192.168.165.93 هستش. (من چون روی یک سیستم دارم بالا میارم هر سه port ها رو به مقادیر مختلفی expose کردم.)خود Zookeeper از چند پورت مختلف استفاده میکند، 2888 برای ارتباط بین لیدر و سایر نودهای quorum ، 3888 برای انتخاب لیدر در quorum، 2181 برای ارتباط client به zookeeper (بروکر های کافکا از طریق این پورت متصل میشوند.)version: "3.9" services: zk1: hostname: zk1 image: zookeeper:3.9.2 ports: - 2181:2181 - 3888:3888 - 2888:2888 environment: - ZOO_MY_ID=1 - ZOO_SERVERS=server.1=0.0.0.0:2888:3888 server.2=192.168.160.93:2882:3882 server.3=192.168.160.93:2883:3883 zk2: hostname: zk2 image: zookeeper:3.9.2 ports: - 2182:2181 - 3882:3888 - 2882:2888 environment: - ZOO_MY_ID=2 - ZOO_SERVERS=server.1=192.168.160.93:2888:3888 server.2=0.0.0.0:2888:3888 server.3=192.168.160.93:2883:3883 zk3: hostname: zk3 image: zookeeper:3.9.2 ports: - 2183:2181 - 3883:3888 - 2883:2888 environment: - ZOO_MY_ID=3 - ZOO_SERVERS=server.1=192.168.160.93:2888:3888 server.2=192.168.160.93:2882:3882 server.3=0.0.0.0:2888:3888تا اینجا کلاستر zookeeper ساختیم، لازمه دست کم یک broker کافکا به این کلاستر اضافه کنیم.نیازمندی های سخت افزاری کافکاوابستگی کافکا تنها به zookeeper است. بهتر است کافکا روی سیستمی بالا بیاید که cpu quad-core 24GB دارد. در صورتی که compression و ssl را فعال داشته باشیم میزان مصرف cpu بسیار زیاد میشود. اما اگر نیاز به امنیت و compression نداریم میتوان مقدار cpu را کم کرد. RAM مورد نیاز به صورت optimal مقدار 6GB است ولی اگر محیطهای production با بار زیاد، حدودا 32GB رم مورد نیاز است. در سطح سیستم عامل بهتر است بر روی سیستم عامل هایی که linux base هستند بالا بیاید. File system های XFS برای کافکا performance بهتری دارد و ssd ها در اینجا مفید نخواهد بود چون کافکا به صورت sequential میخواند و ssd اصلا نمیتواند بهبودی حاصل کند.کانفیگ های نرم افزاری کافکا، در سطح سیستم عامل، file descriptor limits، max socket buffer و max number of memory map مهم است.(این ها فارغ از بحث container است.) در سمت producer لازم است کانفیگ های acks، compression و batch size در نظر گرفته شود. در سمت consumer نیز بحث fetch size مهم است.کلاستر Kafkaتا اینجا کلاستر zookeeper ساختیم، لازمه دست کم یک broker کافکا به این کلاستر اضافه کنیم. قبلتر نحوه بالا آوردن کافکا رو دیده بودیم که چون zookeeper ها کلاستر شدن، باید آدرس همه نودهاشو به بروکر کافکا در غالب متغیر KAFKA_ZOOKEEPER_CONNECT معرفی کنیم.( کاما separate و بدون هیچ space ای میذاریم)kafka1: image: confluentinc/cp-kafka restart: always depends_on: - zk1 - zk2 - zk3 ports: - 9092:9092 environment: - KAFKA_ZOOKEEPER_CONNECT=192.168.160.93:2181, 192.168.160.93:2182, 192.168.160.93:2183 - KAFKA_LISTENERS=PLAINTEXT://:9092 - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.160.93:9092ما وقتی چند تا بروکر کافکا داریم در کافکا client تنها آدرس یکی را بگوییم کفایت میکند چون از طریق bootstrap server بقیه را میشناسد ولی در مورد zookeeper لازم است حتما آدرس همه نود ها را داشته باشیم.لازمه فایل بالا اومدن بروکر های کافکا رو مجزا از نودهای zookeeper قرار بدید که ابتدا zookeeper رو کانفیگ کرده و در حد 2 3 دقیقه زمان بدیم که zookeeper ها به هم متصل بشن بعدش بروکرهای کافکا رو بالا بیاریم.در این پست به موارد و جزییات بیشتری از نحوه کلاستر شدن zookeeper و kafka و نیازمندی های سخت افزاری و نرم افزاری اشاره کردیم.از اینکه وقت خود را در اختیار من قرار دادین بسیار سپاس گذارم. امیدوارم این مطالب برای شما موثر و مفید باشد. من تلاش کردم مطلب رو به ساده ترین حالت بیان کنم. در صورتی که پیشنهاد یا انتقادی در مورد نحوه ی نگارش و مطالب دارین خوشحال میشم با من در میان بذارین و هر سوال و ابهامی در مورد کافکا داشتین رو در کامنت ها بذارین که بتونم کمک کنم.

Apache-Kafka-بخش چهارم-(نحوه مصرف شدن پیام ها و نیازمندی های اجرا)

مریم محمدی — Sun, 11 Aug 2024 14:00:53 +0330

ما تا الان نحوه کار کلی کافکا، دمویی از اون و نصبش و جزییات بیشتری از نحوه ذخیره سازی پیغام ها توسط کافکا رو با هم مشاهده کردیم. در این نوشتار میخوایم موارد پیشرفته ی دیگه ای مانند نحوه تنظیم order و تضمین مصرف شدن پیغام توسط consumer رو با هم ببینیم. این جزییات برای مصارفی که در اون ترتیب دریافت پیام و گارانتی پردازش همه پیام ها برایشان ضرورت دارد بسیار کاربردی است.در بخش های قبل دیدیم که هر تاپیک به یک یا چند پارتیشن تقسیم میشود. یک نکته ای را در مورد پارتیشن ها در نظر بگیریم هم این هست که هر پارتیشن به حداکثر یک consumer میرسد. یعنی یک consumer میتواند از تعدادی partition بخواند ولی دو consumer مختلف نمیتوانند از یک partition پیغام بگیرند. اضافه کردن consumer ها برای افزایش بار پردازشی است، چون به هر پارتیشن نهایت یک consumer متصل میشود، بنابراین برای زیاد شدن بار پردازشی لازم است متناسب با افزایش تعداد پارتیشن ها، تعداد consumer ها را نیز زیاد کرد.Message Orderپیغام های تولید شده برای هر تاپیک، براساس انتخاب بروکر به partition ای داخل آن تاپیک وارد میشود. نحوه این انتخاب پارتیشن ها، با الگوریتم های قابل تغییر انجام میشود. برای مثال نحوه نوشتن پیغام بر روی پارتیشن ها به صورت دیفالت، الگوریتم round robin است. کافکا key پیغام ها را hash کرده و بعد با استفاده از این hash و الگوریتم خود، شماره پارتیشن برای ذخیره سازی داده ها را به دست می آورد. کلیدهایی که یکسان هستند چون hash آنها ثابت است همگی در یک پارتیشن ذخیره سازی میشوند. بتابراین اگر order خواندن پیغام توسط consumer برای ما اهمیت داشته باشد، لازم است برای پیغام های تولیدی key یکسانی قرار دهیم.Consumer Groupدر کافکا مفهومی به اسم consumer group وجود دارد که consumer هایی که میخواهند از یک topic مشترکا خوانده و پردازش کنند، با هم تشکیل یک consumer group را میدهند. بیشتر برای اینکه توان پردازشی پیام ها بالا برود این consumer group تشکیل میشود. نحوه هماهنگی پردازش پیغام های تاپیک، تحت عنوان consumer group انجام میشود. تشکیل consumer group با دادن id یکسان برای گروه به consumer ها تشکیل میشود. ایجاد consumer group برای افزایش بار پردازشی پیام ها ضروری است ولی در صورتی که یک consumer از دست برود و یا consumer دیگری بنابر نیاز اضافه شود لازم است به نحوی کافکا بداند که تا الان چه پیغام هایی توسط consumer ها خوانده شده است که با اضافه و کم شدن consumer ها پیغام تکراری ارسال نکند. این کار برای کافکا با استفاده از متغیری به نام offset commit انجام میشود که در ادامه نحوه ی کار آن را میبینیم.Offset Commitبا خواندن پیام ها از یک تاپیک توسط consumer، باعث میشود که offset هایی که توسط هر consumer خوانده شده است به کافکا از consumer ارسال گردد. کافکا به دلیل اینکه پیغامی را چند باره به consumer ارسال نکند و یا در صورت از دست رفتن consumer و اضافه شدن consumer جدید بتوان پردازش از سر گرفته شود این آفست ها را نگهداری میکند که با نام offset commit شناسایی میشود. کافکا اینها را در یک تاپیک سیستمی به نام __consumer_offsets برای هر consumer ای ذخیره میکند (برای consumer های داخل consumer group اینها برای گروه نگهداری میشود.) این ارسال آفست خوانده شده از سمت consumer تحت درخواستی به اسم offsetCommitRequest انجام میشود. بنابراین acknowledge در بحث consumer ها با استفاده از همین consumer offset است.گاهی ممکن است بین این offset کامیت شده با آخرین پیام خوانده شده فاصله ای باشد چون این ارسال committed offset برای افزایش throughput به صورت asynch است و بسته به پیاده سازی library کافکا ممکن است مثلا بعد از خواندن یک batch کامل مقدار last commited offset به روز رسانی شود.در هر پارتیشن دو تا offset دیگه هم هست به نام های log end offset که نشان میدهد offset آخرین پیامی که produce شده چیست و high watermark به آفست ای اشاره میکند که replicate شده و در اینجا نیز یک تاخیری بین پیغام های produce شده و high watermark وجود دارد. Kafka Lagsیک سری key ها در کافکا وجود دارد که KPI هستند یعنی برای بررسی performance سیستم streaming بکار می روند. Consumer-Lag یکی از اینها است که تفاضل زمانی آخرین داده ای که خوانده شده و آخرین داده ای که produce شده است، میباشد.Lag = Last commited Offset – Last Produced Messageاین lag باید در کمترین مقدار نگهداری شود و برای مانیتور کردن سیستم لازم است مقدار این lag بررسی گردد. این lag در تاپیک سیستمی __consumer_offset توسط کافکا ذخیره میشود. این lag لازم است هواره کمینه باشد زیرا پیغام ها در کافکا based on retention هستند یعنی نهایتا تا یه بازه زمانی در کافکا ذخیره میشوند. (از دست نمیرن ولی فقط در آن بازه ی زمانی که قابل تغییر است ذخیره میشوند.)Monitoring Kafkaکافکا را به چند طریق میتوان مانیتور کرد، یکی از ابزارها Burrow است که در کنار Lag تایم، نرخ produce پیغام و نرخ consume کردن پیغام را میتوان با آن مانیتور کرد و اگر تاخیر داریم متوجه بشیم کدام قسمت از سامانه مشکل دارد. burrow میتواند نمودارها را time-base نیز نشان دهد که lag را در زمان های مختلف نشان میدهد. نمودارهای partition-base نیز داریم که گاهی میتوان علت مشکل را ناشی از تنها یک پارتیشن باشد را تشخیص داد. یکی دیگر از مقادیری که میتوان مانیتور کرد lag به نرخ تولید پیام است که در ارزیابی ها بسیار مهم است.در بحث consume پیام ها، مفهوم over consumption و under consumption داریم. این مفهوم به این اشاره میکند که گاهی پیش می آید که ما یک پیام را اصلا consume نکنیم ولی بعضی پیام ها را چندین بار consume کنیم. فرض کنیم consumer از دست میرود و commit offset ها دیر اتفاق افتاده باشد در این حالت با برگشت مجدد consumer، پیام های تکراری ارسال میگردد که over consumption است. Under consumption زمانی اتفاق می افتد که offset از دست برود برای مثال consumer تنظیم کند پیغام latest همواره ارسال گردد که در این حالت با از دست رفتن کوتاه مدت consumer یک سری پیغام ها پردازش نمیشود یا consumer دیگری اضافه شود و offset را reset کند.در این پست به موارد و جزییات بیشتری از نحوه مصرف شدن داده ها توسط consumer ها و نیازمندی های سخت افزاری و نرم افزاری اشاره کردیم و برخی کانفیگ های مهم رو بررسی کردیم.از اینکه وقت خود را در اختیار من قرار دادین بسیار سپاس گذارم. امیدوارم این مطالب برای شما موثر و مفید باشد. من تلاش کردم مطلب رو به ساده ترین حالت بیان کنم. در صورتی که پیشنهاد یا انتقادی در مورد نحوه ی نگارش و مطالب دارین خوشحال میشم با من در میان بذارین و هر سوال و ابهامی در مورد کافکا داشتین رو در کامنت ها بذارین که بتونم کمک کنم.

Apache-Kafka-بخش سوم (جزییات پیام ها)

مریم محمدی — Wed, 24 Jul 2024 13:03:39 +0330

ما تا الان نحوه کار کلی کافکا، دموی کوچیکی از اون و نصبش رو با هم مشاهده کردیم. در این نوشتار میخوایم موارد پیشرفته ی کافکا که شامل متادیتا و جزییات پیام ها، همچنین ذخیره سازی و دریافت پیغامها توسط کافکا رو بیشتر بدونیم. این جزییات برای مصارفی که از دست نرفتن حتی یک پیغام هم حیاتی است بسیار لازم است.Meta Dataدر کافکا بستر متادیتا با استفاده از سه بخش key ،value و header فراهم شده است. بخش header اجباری است ولی key، value ها اختیاری هستند. اغلب توصیه میشود مستقیم درگیر هدرها و ارتباط با درایور کافکا نشوید. ولی برای مثال در کاربردهایی مانند audit پیام(پیام رسیده باشد و state کنونی در سامانه چیست) و یا track پیام ها، از هدر استفاده میشود. برای مثال id پیام را در هدر قرار میدهیم و به راحتی track پیام را در جاهای مختلف داشت. در مبحث پیام رسان ها هم اغلب ادعا میشود ما end to end encryption داریم و اگر نخواهند درگیر verify کردن sign با باز کردن پیام شوند میتوانند اطلاعاتی را با استفاده از هدرها منتقل کنند.Serializationکافکا به صورت دیفالت تایپ های String، Double، Long، Integer و Byte رو خودش serialize و deserialize میکند. اما در داخل خود پلتفرم چه در مورد متا دیتا و چه پیغام ها، تایپ داده به byte array تبدیل میشود. اگه بخوایم دستی این سریالایز، دی سریالایز رو انجام ندیم باید سریالایزر custom بنویسیم. سریالایزر تعریف کردن یه کار اضافه ای هم انجام میده که ما اسکیما رو دستی چک نمیکنیم و خود پلتفرم بررسی میکنه که داده ها اون تایپ رو داشته باشن. اگر یه وقت بخوایم compression داشته باشیم یا sign و encryption انجام بدیم باید در بخش سریالایزیشن این رو ببینیم.Replication Factorما تا اینجا تنها یک بروکر استفاده کردیم. در صورتی که این بروکر به هر علتی مشکلی داشته باشد producer ها دیگر قادر به ارسال پیام نیستند و پیام های ارسالی تا کنون نیز اگر به دست consumer ها نرسیده باشد از بین رفته است. بنابراین در صورت داشتن تک بروکر، در مقابل مشکل خرابی بروکر تحمل پذیر نیستیم و در صورت داشتن تنها یک بروکر، لازم است بهای از دست رفتن پیغام ها و downtime را بپردازیم. برای پیش نیامدن اینطور مشکلاتی، کافکا راه حل استفاده از چند بروکر را دارد. مدیریت وجود چند بروکر در کافکا را zookeeper بر عهده دارد. بروکر ها خود را در zookeeper رجیستر کرده و از طریق آن وجود باقی بروکر ها را متوجه میشوند. این تنها کاری است که zookeeper در کافکا انجام میدهد.در صورتی که ما تنها یک بروکر داشته باشیم یا داده ها را تنها در یک بروکر ذخیره کنیم، ممکن است data loss اتفاق بیافتد. برای همین کافکا ویژگی دیگری به نام replication factor را ارائه داده است.برای جلوگیری از fail نشدن و از دست نرفتن پیغام زمانی که broker غیرفعال شود، هر پیغام به تعدادی broker همزمان ارسال میشود. به این تعداد کپی از داده ها replication factor گفته میشود.(اصل داده به بروکر leader میرسد و از آنجا در سایر بروکر ها ذخیره میشود.) در زمان تعریف topic، مقدار replication مورد نیاز نیز قابل تنظیم است. با تنظیم این عدد به این تعداد کپی بر روی بروکر های متفاوت از این داده ذخیره میشوند. بنابراین لازم است همزمان به تعداد replication factor بروکر در دسترسی باشد. داده ها بر روی چندین بروکر ذخیره میشود ولی producer تنها با یک بروکر در ارتباط است که به آن partition leader میگویند.نحوه انتخاب این election به این صورت است که zookeeper از بروکرها میپرسد چه کسی حاضر است لیدر باشد و همه بروکرها درخواست لیدر شدن خود را ارسال میکنند و هر کدام از درخواست ها که سریع تر به zookeeper برسد به عنوان لیدر انتخاب میشود! در ابتدای کار با کافکا یک سری configuration تنظیم میکنیم و برنامه برای تولید/مصرف پیغام بالا می آید، پیش از ارسال داده یا subscribe شدن consumer، یک meta data رد و بدل میشود. در این متادیتا تعداد replication ها، protocol تبادلی، زمان های insync، لگ، leader هر پارتیشن ، تعداد پارتیشن ها و ... مشخص میشود. بنابراین در ابتدا producer و consumer میداند leader کیست. پیغام تنها برای leader ارسال میشود و این بروکرهای داخل کلاستر هستند که با هم تعامل دارند تا داده را بین خود نیز تبادل کنند. بروکر ها به partition leader درخواست یا issue ای برای fetch data ارسال میکند تا کپی ای از آن partition داشته باشند. کافکا لازم است به نحوی بفهمد که پراسس داده ها تمام شده است، اینکار با دریافت پیغام offset commit انجام میشود.در صورتی که leader از دست برود، یک تاخیر چند میلی ثانیه ای ایجاد میشود و بعد از آن بین بروکر ها یک election (مثل دفعه قبل) انجام میشود و لیدر بعدی انتخاب میشود و دوباره این leader در غالب متا دیتا به همه معرفی میشود. isr (In Sync Replica) تعداد بروکر هایی که داده هایشان با هم synch میباشد است. این بروکر ها با یک lag ای سینک میشوند که قابل تنظیم میباشد. (replica lag time max ms) پیش فرض این لگ تایم 30 ثانیه میباشد. leader ها همواره لیستی از بروکرهای in sync replica را دارند. هر بروکر ماکسیمم طی بازه لگ تایم باید sync شود و اگر بیشتر از این خود را آپدیت نکرده باشد، out of sync در نظر گرفته میشود. Message Persistenceدر کافکا مسئله دیگری که مهم است delivery و durability است. در مبحث durability باید براساس بیزینس و مسئله تصمیم گیری کرد. گاهی مهم است که تمامی پیام هایی که ارسال میشوند، پردازش شود. ولی در بعضی مسائل اگر 90 درصد پیغام ها پردازش شوند نیز کفایت میکند و از دست دادن 10 درصد باقی مانده مشکلی ایجاد نمیکند. بنابراین بیزینس مسئله تعیین کننده ی durability است. پیغام از جانب producer تولید و در بروکر قرار میگیرد. بروکر تضمین میکند که براساس کانفیگ های انجام شده، پیغام به دست consumer میرسد. این تضمین با کمک acks (acknowledge) انجام میشود. acks سه مقدار متفاوت none، one و all دارد. برای مثال در صورتی که بیزینس track رفتار کاربران باشد دریافت تک تک این کلیک ها کم اهمیت است ولی اگر کافکا را برای transaction های بانکی به کار ببریم اطمینان از دریافت آنها بسیار مهم است. برای همین در producer متغیری به نام acks وجود دارد که در صورتی که "acks" = none، از جانب بروکر ack ای مبنی بر دریافت پیغام به producer ارسال نمیکند و producer هر پیغامی که ارسال میکند را دریافت شده فرض میکند حتی اگر بروکر ای برای دریافت نباشد. بنابراین در چنین حالتی ممکن است data loss اتفاق بیافتد.اگر مقدار acks = 1 باشد، بروکر با دریافت پیغام یک ack به producer ارسال میکند. بنابراین در این حالت producer متوجه میشود که پیغام به درستی دریافت شده است. به محض دریافت و ذخیره ی پیغام توسط یکی از بروکرها، بروکر ack را به producer ارسال میکند ولی هنوز commit فرض نشده است. بعد از نوشته شدن داده توسط تمامی بروکرهای دیگر داده commit شده فرض میشود.در این حالت پیش از commit بودن و ذخیره ی داده در تمامی بروکرها، ممکن است داده در partition leader ذخیره شود (ولی هنوز در سایر بروکرها ذخیره نشده باشد) و با از دست رفتن بروکر، داده نیز از دست میرود. برای جلوگیری از چنین مشکلی، acks = all میشود و مادامی که همه ی بروکرها داده را ذخیره نکرده باشند، Producer پیغام ACK را دریافت نمیکند. بنابراین اگر مقدار acks = all باشد، تمامی بروکر هایی که وجود دارند باید تضمین کنند که داده را دریافت و ذخیره کرده اند. در حالت acks = all کافکا برای ارسال ack، متغیر دیگری را نیز لحاظ میکند. این متغیر، min.insynch.replica است. به صورت دیفالت این مقدار 1 است. کافکا در حالت acks = all رفتار دیگری نیز دارد، برای مثال یکی از بروکر ها ارتباطش را از دست داده است و داده ها را ذخیره نمیکند. اگر min.insynch.replica کمتر از replication factor باشد، کپی داده به تعداد min برای ارسال ack کفایت میکند. این رفتار عجیب کافکا برای اینکه در خیلی از مواقع ارتباط نتورکی بروکر ها از دست میرود کارایی دارد که تولید پیغام را متوقف نکند.در صورتی که به اندازه ی min.insynch.replica بروکر فعال وجود نداشته باشد، producer پیغام خطا دریافت میکند.در این پست به موارد و جزییات بیشتری از کافکا اشاره کردیم و برخی کانفیگ های مهم برای ارسال بدون خطای داده ها را بررسی کردیم. نحوه ی نگهداری داده ها بررسی شد و نحوه ی کانفیگ چند بروکر ار دیدیم. برای یک برنامه نویس دانستن این حدود از کافکا کفایت میکند و برخی مطالب اشاره نشده در این مطالب مگر در موارد خاص یا به عنوان admin کافکا برایتان مفید باشد.از اینکه وقت خود را در اختیار من قرار دادین بسیار سپاس گذارم. امیدوارم این مطالب برای شما موثر و مفید باشد. من تلاش کردم مطلب رو به ساده ترین حالت بیان کنم. در صورتی که پیشنهاد یا انتقادی در مورد نحوه ی نگارش و مطالب دارین خوشحال میشم با من در میان بذارین و هر سوال و ابهامی در مورد کافکا داشتین رو در کامنت ها بذارین که بتونم کمک کنم.

Apache-Kafka-بخش دوم

مریم محمدی — Wed, 24 Jul 2024 08:48:11 +0330

در بخش های اول و نصب، معماری کلی و نحوه ی نصب کافکا بیان شد. در این بخش دمویی از انتقال پیام بین consumer و producer ها و سپس مفاهیم عمیقتر و معماری جزیی تری از کافکا آمده است.Kafka Client Libraryدر ابتدا کافکا تنها برای جاوا و scala قابل استفاده بود ولی اکنون اغلب زبان های محبوب مانند C/C++، Go، Java،، .NET، Python و Scala در کنار سایر زبان هایی مانند Erlang, Groovy, Haskell, Kotlin, Lua, Node.js, OCaml, PHP, Ruby, Rust, Tcl, Swift قابلیت استفاده از کافکا را دارند.برای نوشتن دموی اولیه از لایبرری Kafka Client استفاده میکنیم. این لایبرری از سه قسمت تشکیل شده است. کارهای administrative مانند ایجاد تاپیک، مربوط به بخش Admin، تولید پیغام مربوط به Producer و مصرف پیغام مربوط به Consumer است.بخش های مختلف لایبرری Kafka-Clientبرای بررسی دقیق ویژگی های موجود در کافکا میتوان به سایت که توسط سازنده های کافکا ایجاد شده است، مراجعه کرد.برای ارسال و دریافت پیغام نیاز به Topic ای برای نگهداری پیام، Producer ای که پیغام را send کند و consumer ای که پبغام را poll(دریافت) کند، داریم.برای این منظور دو/سه برنامه مینویسیم. من با استفاده از زبان جاوا ورژن 11 این برنامه ها را نوشته ام. برای ساخت Topic میتوان با استفاده از CLI، Kafka tools و یا کد نویسی استفاده شود. تجربه من در شرکتم این هست که تاپیک ها با استفاده از kafka tools در هنگام عملیاتی شدن ساخته میشوند. در همه حالتها نیز لازم است، آدرس بروکر را در غالب متغیر bootstrap.server داشته باشیم. در صورت استفاده از CLI میتوان بعد از بالا بودن کافکا با استفاده از دستور زیر تاپیک مورد نظر را ساخت.kafka-topics.sh --create --topic --partitions --replication-factor --zookeeper در اینجا نام تاپیک مورد نظر، اندازه ی partition و سایر متغیر ها نیز به ترتیب تعداد replication مورد نیاز و آدرس zookeeper و port ای روی آن بالا آمده است، میباشد. در صورت استفاده از kafka tools که به نظرم برای این کار بهترین ابزار میباشد، ابتدا kafka-tools را نصب کرده سپس لازم است یک connection جدید با استفاده از انتخاب Add new connection از منوی File ساخته شود. نحوه ساخت connection به صورت زیر است.نحوه ساخت connection در kafka toolsدر اینجا لازم است آدرس zookeeper، پورت آن و نامی برای کلاستر انتخاب شود. بعد از اتصال به آن در قسمت topic میتوان تمامی تاپیک های ایجاد شده را مشاهده کرد و یا تاپیک جدیدی را ایجاد کرد.نحوه ایجاد تاپیک با استفاده از kafka-toolsدر منوی سمت چپ از cluster اضافه شده، ابتدا لیست بروکر ها(در اینجا من براساس نصب اشاره کردم که تنها یک بروکر را بالا می آورم که آدرس و پورت آن آمده است) سپس لیست تاپیک ها( که در اینجا خالی است) آمده است که با right click بر روی Topics و انتخاب گزینه ی create topics پنجره نشان داده میشود(شکل بالا). نام topic، تعداد replication ها از آن و مقدار partition نیز قابل مقداردهی میباشد و با گزینه ی Add، تاپیک مورد نظر اضافه میشود. حالت سومی نیز برای ایجاد تاپیک وجود دارد که با استفاده از کدنویسی است. واقعیت من چون تجربه کمی در کافکا دارم تا کنون ندیده ام با کدنویسی تاپیک نیز ساخته شود در این صورت کد زیر قابل استفاده است. ابتدا لازم است Admin کانفیگ شود که برای اینکار آدرس بروکر را قرار میدهیم . سپس با استفاده از متد createTopic از admin یک تاپیک ایجاد میکنیم. برای ساخت تاپیک فعلا نیاز به نام آن تنها داریم و بقیه ی موارد آن را خالی قرار میدهیم. من در اینجا یک تاپیک با نام "my_topic1" ایجاد کردم. چون کافکا را بر روی سیستم خود تنها اجرا کرده ام آدرس بروکر نیز localhost است که بر روی پورت 9092 قابل استفاده است. کد این قسمت در زیر آمده است.public static void main(String[] args) { Admin admin = Admin.create(Map.of("bootstrap.servers", "localhost:9092")); admin.createTopics(Arrays.asList(new NewTopic("my_topic1", Optional.empty(), Optional.empty()))); }تا الان ما تاپیک را ایجاد کردیم در مرحله ی بعد لازمه پیغام را produce کرده به topic ایجاد شده ارسال کنیم. در اینجا کدهای لازم producer و consumer با استفاده از جاوا آمده است ولی در صورتی که بر بستر اسپرینگ کار میکنین spring kafka بسیار ساده تر است که در این نوشتار به اون نمیپردازیم. پیغام با استفاده از ProducerRecord تولید میشه و از طریق KafkaProducer ارسال میشه. حالا برنامه ی دیگه ای مینویسیم که پیغام را تولید و ارسال کند. برای تولید پیغام نام تاپیک و محتوای پیغام کافی است. ایجاد KafkaProducer نیاز به آدرس بروکر، Serializer برای key و value دارد. پیغام ایجاد شده نیز توسط متد send از KafkaProducer به تاپیک ارسال میشود. در انتها نیز برای آنکه پیغام های buffer شده یکجا ارسال شود، آن را flush میکنیم. در ادامه دلیل وجود serializer ها و فلاش را میگم. با اجرای برنامه پیام "Hi hi" در تاپیک قرار میگیرد. کد این قسمت از برنامه در زیر آمده است.public static void main(String[] args) { Properties prop = new Properties(); prop.put("bootstrap.servers", "localhost:9092"); prop.put("key.serializer", StringSerializer.class); prop.put("value.serializer", StringSerializer.class); KafkaProducer kafkaProducer = new KafkaProducer(prop); ProducerRecord producerRecord1 = new ProducerRecord("my_topic1", "Hi hi"); kafkaProducer.send(producerRecord1); kafkaProducer.flush(); kafkaProducer.close(); }تا اینجا تاپیک و تولید کننده ی پیغام را ساخته ایم. در ادامه برنامه ی دیگری مینویسیم که Consumer را ساخته و به تاپیک subscribe کند. ساخت Consumer با استفاده از کلاس KafkaConsumer انجام میشود که به آدرس بروکر، نام گروهی که به آن تعلق دارد و desrializer برای deserialize شدن key و value نیاز دارد و از طریق متد subscribe به تاپیک مورد نظر subscribe شده. Fetch کردن دیتا از تاپیک (یا پارتیشن خاصی در تاپیک) با استفاده از متد poll است. پیغام در غالب کلاس ConsumerRecord دریافت میشود. هر پیغام حاوی value یا مقدار پیغام، شماره ی پارتیشن(partition)، شماره آفست (offset) است. ما در این برنامه فقط میخواهیم همه پیغام های رسیده را چاپ کنیم. چون میخواهیم هر پیغامی که آمده باشد را بگیریم، انتظار دریافت پیغام را در یک loop قرار داده ایم، بنابراین کد زیر را اجرا کرده و منتظر ارسال پیغام میمانیم. سپس با اجرای برنامه ی تولید پیغام، پیغام ها را به تاپیک ارسال میکنیم و میبینیم که پیغام بلافاصله به دست consumer میرسد.public static void main(String[] args) { String consumerGroup = "dispatch-service" Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", consumerGroup); props.put("key.deserializer", StringDeserializer.class); props.put("value.deserializer", StringDeserializer.class); KafkaConsumer kafkaConsumer = new KafkaConsumer(props); kafkaConsumer.subscribe(List.of("my_topic1")); while (true) { ConsumerRecords records = kafkaConsumer.poll ( Duration.ofMillis ( 1000 )); for (ConsumerRecord record : records) { System.out.printf("Message %s at offset %d of partition %d%n", record.value(), record.offset(), record.partition()); } } }با اجرای برنامه ها به ترتیب تولید تاپیک، ساخت consumer و ساخت producer، خروجی زیر که نشان دهنده ی رسیدن پیغام به consumer است در کنسول consumer چاپ میشود.کافکا ترتیب پیغام ها را در هر partition تضمین میکند. بنابراین اگر لازم است پیغامی به ترتیب به دست مصرف کننده برسد لازم است key آنها یکسان باشد.Timeout and Reties:در صورتی که بروکر بالا نباشد و producer پیغام را ارسال کند و ack از جانب بروکر به دست producer نرسیده باشد(در صورت فعال بودن)، producer فرض میکند پیغام به دست بروکر نرسیده یا خطایی پیش آمده، به همین دلیل به صورت اتوماتیک پیغام را مجدد ارسال میکند. این تلاش مجدد به تعداد retries انجام میشود که در غالب متغیری به نام "retries" قابل تنظیم است. در کنار این retries زمان کل ارسال پیغام از زمانی که با متد send ارسال شود تا ack به دست producer برسد یک Timeout دارد که آن نیز با استفاده از متغیر "delivery.timeout.ms" قابل تنظیم است. در صورتی که timeout.ms سپری شود فارغ از اینکه چه تعداد retries برای ارسال پیغام انجام شده، پیغام ارسال نشده فرض میشود. برای مثال حالتی را در نظر بگیرید که retries = 5 و timeout =5ms باشد و زمان 5ms از send شدن پیغام گذشته باشد در حالی که تنها 2 بار تلاش مجدد برای ارسال پیغام صورت گرفته باشد و ack فعال باشد، پیغام ارسال نشده فرض میشود و خطا برمیگردد. به صورت دیفالت timeout= 2 minutes و retires=Integer.MAX_VALUE است. یعنی به صورت دیفالت تا زمانی که timeout نرسد، producer به هر تعداد مجدد برای ارسال پیغام تلاش میکند. Producer برای همه ی خطاها مجدد ارسال نمیکند برای مثال اگر serializer اشتباه ست شده باشد ارسال مجدد تاثیری ندارد و برای این خطا پیغام مجدد ارسال نمیشود.در کافکا producer برای هر retry به اندازه ی retry.backoff.ms صبر کرده و سپس تلاش بعدی را انجام میدهد. این زمان جلوی ارسال مجدد و دائم در یک چرخه را میگیرد.در کنار این مفهوم و متغیرهای زمانی، کافکا متغیر دیگری به نام "max.in.flight.requests.per.connection" نیز دارد. این متغیر به معنی حداکثر تعداد ارسال های unacknowledged (بدون دریافت ack برای آن) است. در صورتی که این مقدار از 1 بزرگتر باشد، به این معنی است که ترتیب قرار گیری پیغام ها در partitionها به همان ترتیب ارسال ممکن است نباشد. برای مثال حالتی را فرض کنید که این متغیر 2 باشد یعنی میتوان دو request در یک ارتباط با بروکر ارسال کرد و برای دومی محدودیتی برای اینکه اولی توسط بروکر ذخیره شده است را نداشت. حال producer این دو درخواست را ارسال میکند و پیغام های درخواست اول fail میشود ولی درخواست دوم در بروکر ذخیره میشود بنابراین پیغام های سری دوم قبل از پیغام های اول در بروکر ذخیره میشود و ترتیب ذخیره سازی پیغام ها به همان ترتیب ارسال نشده است.Producer Flowدر demo بالا، پیغام بعد از ارسال flush نیز شده.kafkaProducer.send(producerRecord1); kafkaProducer.flush();برای علت اینکار لازمه flow ارسال پیغام توسط producer را با جزییات بیشتر بررسی کنیم.روند ارسال پیغام توسط producerپیغام های تولید شده توسط Producer، در قدم اول توسط serializer، سریالایز میشوند. سپس partition ای که لازم است پیغام ها در آن نگهداری شوند مشخص میشود. کافکا چون میخواد سرعت انتقال بالایی داشته باشه، پیغام ها را به صورت batch و دسته ای ارسال میکنه. چون ارسال هر پیغام به تنهایی سربار زیادی داره. بنابراین به ازای پیغام های هر پارتیشن، آن ها را در بافری نگهداری میکنه و بعد همه را با هم به آن پارتیشن میفرسته. پیغام در buffer ارسال پیغام برای آن پارتیشن ذخیره میشه.در نهایت پیغام های buffer شده به صورت batch به کافکا broker ارسال میشود. سایز batch توسط متغیر "batch.size" قابل تغییره. در صورت محدودیت حافظه این مقدار را کاهش داده و برای افزایش نرخ انتقال نیز میتوان مقدار آن را افزایش داد. راه دیگه ای که فارغ از حجم و بسته به زمان هست با استفاده از متغیر "linger.ms" انجام میشود. که در این حالت پیغام ها در صورت پر شدن batch یا رسیدن تایم ست شده ارسال میشوند. مقدار دیفالت این متغیر 0 است.Compressionدر کنار serialize کردن پیغام ها، میتوان آنها را compress و حجم آنها را کاهش داد. برای اینکار لازم است الگوریتمی که برای compress به کار میرود را با استفاده از متغیر "compression.type" ست کرد. با ست شدن این متغیر، کافکا هر پیغامی که در buffer نگهداری میکند و هم batch ارسالی را compress میکند. کافکا اینکار را تنها برای سمت producer انجام میدهد و سمت consumer بی تغییر باقی میماند. کاهش حجم باعث افزایش بار بر روی cpu و زمان برای compress شدن دارد ولی چون سربار cpu برای تولید پیغام بالا نیست اغلب پیغام ها به صورت compress ارسال میشود.Consumersدر بخش قبل گفتیم consumer ها offset پیغام دریافتی را برای بروکر تاپیک ارسال میکنند. در ادامه جزییات بیشتری از نحوه ی تعامل consumer با کافکا آمده است.اولین consumer ای که به کافکا متصل میشود، کافکا یک بروکر به آن اختصاص میدهد که اصطلاحا Group Coordinator برای consumer های یک گروه نامیده میشود. این coordinator لیست تمامی consumer های همگروه و متصل را نگه میدارد و با اضافه یا حذف شدن consumer ها این لیست را به روز میکند.زمانی که یک consumer حذف میشود، این coordinator کار partition rebalance را انجام میدهد. برای partition rebalance ابتدا تمامی تخصیص های انجام شده برای پارتیشن ها از بین میرود.(به این partition revoke میگویند) سپس coordinator لیست consumerها را به یکی از consumer ها به نام consumer g leader میفرستند. (Consumer leader هم اولین consumer ای هست که گروه را ایجاد کرده)این consumer مسئول اختصاص partition به consumerهای همگروه است و بعد لیست این partitionهای تخصیص یافته را برای coordinator باز میگرداند.این coordinator نیز براساس لیست بازگشته، partitionها را تخصیص میدهد.در مورد rebalance شدن لازمه به این نکات توجه بشه:باعث ایجاد gap ای بین پروسه ها میشه.پارتیشن ownership یا consumerای که مصرف کننده ی پیغام های یک partition بود ممکن است عوض بشه.هر خطایی که باعث fail شدن اپلیکیشن مصرف کننده شود لازم است close روی آن صدا زده شود تا coordinator سریعتر partition rebalance را انجام دهد. گاهی این مشکل خطای فیزیکی است و سرور نتوانسته به کافکا خبر دهد. در این حالت coordinator باید به نحوی این را متوجه شود. روش کافکا برای اینکار heart beat است. consumer ها هر چند مدت یکبار یک پیغامی مبنی بر سالم بودن به coordinator ارسال کند. زمان بین ارسال ها توسط متغیری به نام "heartbeat.interval.ms" قابل تنظیم است. در کنار این interval ها مدت زمانی نیز لازم است به عنوان timeout لحاظ شود که بعد از عدم ارسال heartbeat، کافکا consumer را از دست رفته فرض کند. این مدت زمان نیز توسط متغیر "session.timeout.ms" تعیین میشود. بنابراین heartbeat.interval لازم است کوچکتر از session.timeout باشد وگرنه coordinator آن را حذف کرده و partition rebalance انجام میدهد.در این پست دموی کوچکی از نحوه ی ارسال و دریافت پیغام ها آمد. نحوه ی کارکرد کافکا برای ارسال و دریافت ها با جزییات بیشتر بیان شد. کافکا سه خصیصه برای خود بیان کرده بود که شامل fault tolerance، high availabilibty و consistency است. بعضی از این خصیصه ها با این موارد تا کنون گفته شده پوشش داده نمیشن برای مثال در صورتی که بروکر fail شود این موارد نقض میشه. در پست سوم ویژگی های دیگه ای از کافکا بررسی میشه که در صورت بروز مشکل در بروکر باز هم ویژگی های اصلی کافکا حفظ بشن.

Apache Kafka-بخش اول

مریم محمدی — Sat, 06 Jul 2024 08:56:06 +0330

من دوست دارم مطالبی که یاد میگیرم را با افراد دیگر نیز مطرح کنم. در حال حاضر یادگیری Kafka را شروع کرده ام و گام به گام مواردی که میخوانم را با شما مطرح میکنم. نوشتار کافکا در چند پست جداگانه ای آمده است.در گام اول کلیاتی از ساختار و معماری کافکا بیان شده است. سپس توضیح کلی از مفاهیمی مانند broker، partition و topic شرح داده شده است. در گام دوم نحوه نصب Kafka و شرح کامل مفاهیم آن و در گام سوم نحوه تولید پیغام توسط producer و مصرف پیغام توسط consumer بیان شده است و در پایان موارد آتی برای استفاده از کافکا آمده است.دلایل به وجود آمدن کافکا ارتباط بین بخش های مختلف یک سیستم به صورت سنتی به حالت request/response است. برای مثال یک سیستم درخواست خودرو را فرض کنین. با ارسال request از مشتری به سیستم درخواست خودرو، ابتدا این سرویس به مسیریاب، درخواست مسیر میدهد، سپس با استفاده از پاسخ آن، درخواست دیگری به سرویس قیمت دهی برای محاسبه ی هزینه ی سفر میدهد، هزینه ی سفر برای کاربر توسط این سرویس حساب شده و به عنوان پاسخ به سرویس booking بازگردانده میشود. سرویس booking نیز هزینه را در غالب response به کاربر نشان داده میشود. با ارسال request دیگری از کاربر مبنی بر تایید هزینه، سیستم request ای به سیستم توزیع کننده ی درخواست(dispatcher service) یا همان خودرو یاب ارسال میکند و خودرو یافت شده این سرویس را در غالب response دیگری برای کاربر ارسال میکند. این روند در شکل زیر نشان داده شده.نمونه ای از سیستم مبتنی بر request/responseاین سیستم های مبتنی بر request/response تا مدتها به علت سادگی در معماری و ساختار به خوبی جواب میداد و اکنون نیز برای سیستم های ساده تر به خوبی جوابگو است ولی در صورتی که ارتباطات بین سرویس های مختلف زیاد شود و سرویس ها با یکدیگر ارتباط دو به دو داشته باشند ما یک ارتباط بسیار پیچیده ای داریم که نگهداری آن در گذر زمان با توجه به افزودن شدن نیاز های بیزینسی و داده های جدید بسیار سخت میشود.ارتباط و به دو در سیستم هایی با سرویسهای متنوع راه حل خلاقانه ای که برای رفع این مشکل، استفاده از نوع دیگری از انتقال پیام، به صورت producer-subscriber است. این راه حل به این صورت میباشد که پیام ها publish می شوند و علاقه مندان به آن نوع پیام خود را subscribe کرده و پیام را مصرف میکنند. ( publisher ها message ها را تولید (produce) کرده و subscriber ها آن ها مصرف (consume) میکنند.) براساس این ایده ی انتقال پیغام کافکا ایجاد شد.کافکاکافکا در سال 2010 در LinkedIn برای هندل کردن حجم بالای user activity ایجاد شده است. سپس در سال 2011 به عنوان پروژه ی open source در اختیار عموم قرار گرفت.کافکا یک سیستم opensource ،distributed و یک پلتفرم streaming است که تبادل داده سریع، مقیاس پذیر(scalable) و قابل اطمینان(reliable) بین بخش های مختلف انجام میدهد. برای streaming پلتفرم بودن لازم است ویژگی های زیر را داشته باشد.بتوان داده برای جریان رکوردها ایجاد یا مصرف کرد.جریان رکوردها را به صورت خطا گریز(fault tolerant) و بادوام نگه داری کند.جریان داده را به محض ایجاد شدن هندل کند.به طور کلی وظیفه ی آن انتقال داده است. در محاسبات به انتقال داده، پیغام رسانی(messaging) میگویند. بنابراین کافکا یک سیستم پیغام رسانی (messaging system) است. وجه تمایز کافکا با سایر سیستم های پیغام رسانی، امکان استفاده از آن در سیستم هایی با حجم داده های بالا که نیاز دارند مقیاس پذیر و مقاوم در مقابل خطا است. همچنین کافکا به علت سرعت بالا، قابلیت پاسخ‌گویی در لحظه یا ارسال پیغام real-time را دارد و به علت نگهداری پیغام ها بر روی دیسک، data loss آن کم است.در سیستم های enterprise بخش های مختلف با rate بالا داده تولید میکنند. بدون استفاده از کافکا، نگهداری این داده ها و انتقال آن به سایر قسمت های مرتبط نیاز به افزودن حافظه، منابع زیاد و جستجو برای یافتن داده برای سایر بخش ها است. در سیستم های بزرگ در گذر زمان داده های متنوعی در اپلیکیشن تولید میشوند که این باعث تغییر در نحوه نگهداری و جستجوها میشود. این تغییرات باعث به وجود آمدن شبکه ای پیچیده از انتقال نقطه به نقطه داده بین بخش های مختلف در اپلیکیشن میشود.در گذر زمان نگهداری و اعمال تغییرات بر روی این مدل سیستم ها را سخت میکند زیرا چند دیتابیس، middle ware های خاص منظوره برای جابجا داده ها، سیستم هایی برای استخراج داده های خاص و لود آن در سایر اپلیکیشن ها و جابجا log ها، به وجود آمده است.شایع ترین مشکلات سیستم های پیچیده، جابجایی داده بین دو RDBM است که نمیتوان به راحتی داده بین دیتابیس های مختلف جابجا کرد. همچنین تغییر در schema یک دیتابیس نیازمند اعمال تغییر روی تمامی دیتابیسهای مرتبط و داده های کپی گرفته شده است. حجم بالای log فایلهایی که لازم است بین اپلیکیشن ها جابجا شود نیز یکی از مشکلات عمده است. به دلیل سختی در نگهداری از چنین سیستم هایی، نحوه انتقال داده از روش های پیشین به روش های نوین تری تبدیل شد. کافکا به عنوان Messaging Systemبه طور کلی کافکا یک Messaging System وtopic based publish-subscribe عمومی است. در اینجا producer ها و consumer ها میتوانند برنامه های نوشته شده ی ما باشند. در کافکا موارد و کلمات مفهومی و پایه ای وجود دارد که تعریف هر کدام از این موارد در ادامه آمده است. در سیستم topic based publish-subscribe پیغام تولید شده توسط publisher یا producer به یک topic خاص مرتبط میشوند و پیغام ها به صورت selective به subscriber یا consumer علاقه مند به همان topic میرسد.(شکل(2) ) Topic ها در حقیقت یک دسته و گروه از پیغام ها هستند. در کافکا پیغام ها توسط تولیدکنندگان برای یکی از این topic ها ارسال میشوند(send میشوند) و مصرف کننده به یکی از این topic ها subscribe میکند و پیغام ها رو دریافت میکند(poll میکند). کافکا محدودیتی در تعداد topic ها برای subscriber ندارد منتها دریافت پیغام از چند تاپیک به علت اینکه ترتیب پیغام ها تنها در یک تاپیک برقرار است، نیازمند دقت و بررسی برای نحوه ی دریافت پیام ها از چند تاپیک است. این مورد در پست های آتی شرح داده میشود.شکل(2) نمونه ای از سیستم topic-based pub-subسیستمی را فرض کنید که در هر ثانیه، 20 پیغام تولید میکند. این پیغام ها را به کافکا ارسال میکند و مصرف کننده تنها میتوان 10 پیغام در هر ثانیه process کند. بنابراین در گذر زمان این پیغام ها ذخیره شده و به صورت آنی به دست کاربر نمیرسد. برای رفع این مشکل تعداد مصرف کنندگان زیاد میکنیم ولی لازم است کافکا پیغام های رسیده به یک مصرف کننده را به مصرف کننده ی دیگری مجدد ارسال نکند، چون اینها با هم برای انجام یک task در یک گروه قرار گرفته اند.افزودن چند مصرف کننده برای real time بودنکافکا برای اینکه ارسال همزمان پیغام های هر تاپیک به چند مصرف کننده را هندل کند، لازم است بداند به هر مصرف کننده چه پیغامی ارسال شده است و بتواند همزمان به چند مصرف کننده پیغام ارسال کند. مصرف کننده به ازای هر دریافت پیغام، یک ACK برای کافکا مبنی بر دریافت پیغام ارسال میکند. بنابراین کافکا میداند که تا کجا برای consumer ارسال کرده است و آن را جایی نگه میدارد و خود پیغام را فارغ از رسیدن و مصرف شدن توسط consumer در صف خود نگه میدارد. راه حل کافکا برای اینکه پیغام مشترک به همه ی consumer های یک task ارسال نکند، هر topic را به یک یا چند partition تقسیم میکند و برای هر consumer پیغام های همان partition را ارسال میکند. هر پارتیشن به صورت یک فولدر نگه داری میشود که درون آن دو فایل وجود دارد. یکی برای Index ها و یکی برای ذخیره سازی داده ها.پارتیشن بودن هر Topicمصرف کنندگانی که یک task یکسان را انجام میدهند، لازم است پیغام های تکراری دریافت نکنند ولی ممکن استconsumer دیگری اضافه شود که لازم باشد تمامی پیغام ها را دریافت کند. برای هندل کردن اینکار کافکا مصرف کنندگان را به صورت دسته ای تقسیم میکند که به آنها consumer group میگویند. در هنگام ایجاد مصرف کننده با استفاده از متغیری به نام group.id، مشخص میشود که کدام یک از consumer ها در یک دسته قرار دارند. در صورتی که consumer دیگری برای انجام task در کنار این consumer ها اضافه شود و کافکا متوجه شود که این consumer برای انجام تسک یکسانی با بقیه است (consumer group یکسان)، لازم است partition جدیدی به آن اختصاص یابد. به این فرآیند partition rebalance گفته میشود. در صورتی که یک consumer نیز crash کند، partition ها بین consumer های دیگر rebalance میشوند.Partition Rebalanceدر صورتی که بیشتر از تعداد partition ها در یک consumer group، مصرف کننده ای اضافه شود، partitionای به آن اختصاص نمی یابد بنابراین استفاده ای از آن نمیشود. برای مثال consumer4 در شکل زیر پیغامی دریافت نمیکند و idle باقی میماند زیرا کافکا گارانتی کرده است که حداکثر یک consumer به یک partition اختصاص یابد. بنابراین تعداد partition ها مشخص میکند چه تعداد consumer ها میتواند scale شود.تعداد consumer بیشتر از تعداد partitionهاهر پیغام تولید شده یک Topic، key، value و partition دارد. پیغام ها براساس شماره partition در یکی از صف های پیغام آن partition قرار میگیرند. پیغام های ذخیره شده در هر partition یک index مخصوص به خود دارند که به آن offset میگویند. کافکا با استفاده از این offset ترتیب پیغام ها را هندل میکند و consumer ها از هر offset ای که تقاضا دهند میتوانند پیغام را بخوانند. پیغام ها در این partition ها یک مدت زمان مشخصی نگهداری میشوند و بعد از آن از این لیست پاک میشوند و هر پیغامی که به کافکا ارسال میشود در نهایت در یکی از partition ها ذخیره میشود.پارتیشن بودن Topicسروری که این پیغام های تولید شده را گرفته، نگهداری کرده و به دست consumer میرساند، broker نام دارد. هر broker شامل یک یا چند topic است. بروکر ها پیغام ها در فایل هایی log میکنند، مثل فولدرها هستن که پیغام ها را به صورت فایلهایی نگهداری میکنن. broker ها مانند سرور هستند. تفاوت broker و سرور در این است که سرور میتواند با آمدن تقاضاهای زیاد overload شود ولی در کافکا به علت اینکه پیغام های هر topic، فارغ از مصرف شدن و رسیدن به دست consumer به مقدار زمان قابل تنظیمی ذخیره میشود و بعد از آن از بین میرود overload نخواهد شد. در کافکا به broker ها worker node یا به اختصار node نیز میگویند.(شکل (3))شکل(3) معماری کلی پیغام رسانی در کافکاتعریف cluster: به زبان ساده، cluster گروهی از broker ها میباشد. (شکل(4))شکل(4) تعریف cluster نکته ی قابل توجه در اینجا این است که کافکا برای ارتباط و interaction بخش های مختلف استفاده میشود و نه به عنوان دیتابیسی برای ذخیره سازی داده ها چون داده ها به طور دائم در کافکا ذخیره نمیشود. برای ذخیره سازی داده ها میتوان سرویسی برای ذخیره سازی داده ها در پایگاه داده داشت که در صورت لزوم و اهمیت داده ها برای مثال داده های بانکی، آنها را در پایگاه داده ذخیره کرد تا در موارد آتی برای همیشه به آنها رفرنس داشت و یا داده ها را بازیابی کرد.در این پست برخی مفاهیم و ویژگی های کافکا و مختصری از معماری آن بیان شد. به نحوه ی نگهداری پیغام ها و نحوه ی توزیع آن بین consumer ها نیز پرداخته شد. در ادامه ابتدا نحوه ی نصب و کانفیگ کافکا و سپس دموی مختصری کافکا آمده است و در نهایت نیز معماری دقیق تری از کافکا و ویژگی های بیشتری از آن بیان میشود.برای مطالب این نوشتار از درس های Getting Started with Apache Kafka و Designing Event-driven Applications Using Apache Kafka Ecosystem از pluralsight و برخی از سایت ها برای توضیح دقیق تر مفاهیم استفاده شده است.

Apache Kafka- نصب و کانفیگ

مریم محمدی — Tue, 02 Jul 2024 10:31:36 +0330

در این بخش به نحوه ی نصب کافکا پرداخته شده است. در ابتدا روشی که ساده است و به صورت مستقیم با استفاده از دانلود باینری پکیج کافکا بیان شده است که بیشتر در استفاده تمرینی و بر روی سیستم شخصی مناسبه (اگر داکر رو هنوز آموزش ندیدید با استفاده از این روش نصب کنید). در بخش دوم نصب به کمک داکر که برای عملیاتی سازی پروژه ها الزامی استفاده هست رو میبینیم. هر دو روش به نسبت به نظرم ساده است.نحوه نصب و کانفیگ Apache Kafka بر روی ویندوزسیستم من ویندوز 10 است. در ادامه نصب کافکا را بر روی ویندوز آمده است. به طور کلی نصب کافکا بسیار ساده است و بر روی ویندوز یا لینوکس تنها از طریق دانلود و extract محتوای آن قابل استفاده میباشد ولی برای نگهداری طولانی مدت به نظر خودم استفاده از کافکا در صورت امکان به کمک داکر ساده تره نسبت به دانلود و بالا آوردن دستی چند سرور برای عملیاتی شدن پروژه. در ادامه نحوه ی نصب آن و سپس فولدرها و محتوای مهم آن آمده است.پیش نیاز نصب کافکابرای نصب کافکا به موارد زیر نیاز است: 1- باینری پکیج کافکا 2- نصب جاوا( Java 8.0 به بالا)نصب کافکاباینری پکیج کافکا را دانلود کرده و آن را extract کنید.مسیری که کافکا در آن extract میکنیم نباید دارای space باشد. برای مثال در Program File قرار نگیرد. به همین راحتی! الان کافکا رو میشه بالا آورد. منتها قبلش لازمه کانفیگش کرد. در ادامه ی نحوه ی کانفیگ آن برای شروع کافکا آمده است.کانفیگ کافکادر فولدر kafka، چند زیرفولدر داریم. فولدر bin حاوی shell script های متفاوت برای انجام کارها است. در میان اینها 4 اسکریپت برای start و stop کافکا و ZooKeeper مهم هستند. شکل (5) این اسکریپتها را نشان میدهد. در صورتی که از سیستم عامل ویندوز استفاده میکنید، معادل این sh. فایلها، bat. فایل در فولدر windows در داخل فولدر bin، نیز وجود دارد که میتوان از آنها استفاده کرد.شکل(5) اسکریپتهای مورد استفاده در کافکافولدر libs تمامی dependency های کافکا را دارد که شامل ZooKeeper نیز میباشد.فولدر config حاوی فایلهای مورد نیاز برای کانفیگ کافکا است. این فولدر دو فایل مهم server.properties و zookeeper.properties دارد که اولی برای تنظیمات broker ها و دومی تنظیمات ZooKeeper است. شکل(6) فایلهای پوشه ی کانفیگ را نشان میدهد.شکل(6) bat. فایلهای مورد استفاده در کافکابرای start کافکا لازم است ابتدا دو متغیر ست شوند. یکی در فایل server.properties و دیگری در فایل zookeeper.properties است. چون کافکا پیغام ها را به صورت file نگهداری میکند(در بخش های دیگه این مورد اومده)، آدرس فولدر نگهداری این پیغام ها در فایل server.properties متغیر log-dirs تنظیم میشود. بنابراین یک فولدر برای محل نگهداری پیغام های ساخته شود و آدرس آن در این فیلد ست شود.(میتونین از آدرس دهی کامل یا نسبی استفاده کنین.) مثلا خودم در فولدر اصلی کافکا یک فولدر به اسم logs ساختم که داخل اون دو تا فولدر دیگه یکی برای محل پیغام های کافکا و یکی برای zookeeper ساختم و به صورت آدرس دهی نسبی اون رو آدرس دادم. (اگر نمیدونین چرا لازمه مشکلی نداره، تو پست های بعدی این ها رو توضیح میدم.) این فیلد در شکل 7 نشان داده شده است. نکته ای که اینجا هستش کافکا بر روی ویندوز از backslash خوشش نمیاد! از slash استفاده کنین برای مقدار دهی ها. من اینجا از آدرس دهی نسبی استفاده کردم.شکل(7) ست کردن محل ذخیره سازی پیغام های کافکاهمچنین آدرسی برای نگهداری لاگ های zookeeper نیز لازم است ست شود. که برای آن نیز لازم است فولدری مجزا ایجاد شود و آدرس آن در dataDir فایل zookeeper.properties مقدار دهی شود. شکل 8 این متغیر را نشان میدهد.شکل(8) ست کردن محل ذخیره سازی پیغام ها برای zookeeperسپس با اجرای zookeeper-server-start.bat ابتدا zookeeper را start کرده و سپس با اجرای kafka-server-start.bat کافکا را start میکنیم. شکل 9 و 10 شروع کافکا را نشان میدهد. برای اجرای هر کدام از اسکریپتهای start به عنوان ورودی نام و آدرس property فایل آن ( به ترتیب آدرس zookeeper.properties و server.properties) را مینویسیم. برای آدرس دهی میتوان از آدرس دهی نسبی نیز استفاده کرد.شکل(9) بالا آوردن کافکاشکل (10) بالا آوردن zookeeperبه همین سادگی کافکا اجرا میشود! کافکا به صورت دیفالت بر روی پورت 9092 بالا می آید.من اولین باری که kafka را بر روی ویندوز اجرا کردم خطایی گرفتم که میگفت ورودی بسیار بزرگ است! علت این خطا ناشی از محدودیت length در cmd است زیرا در فایل kafka-run- class.bat تعدادی concat به CLASSPATH انجام میدهد که برای رفع آن لازم بود خط زیر در فایل kafka-run- class.batrem Classpath addition for releasefor %%i in ("%BASE_DIR%\libs\*") do ( call :concat "%%i")با خط زیر جابجا شود:rem Classpath addition for releasecall :concat "%BASE_DIR%\libs\*;"نحوه نصب و کانفیگ Apache Kafka با استفاده از dockerمن از قبل داکر رو روی سیستمم که ویندوزی هست نصب کرده بودم. فرض کنیم طرز کار داکر رو همه مون تا حدودی میدونیم بنابراین برای نصب کافکا با کمک داکر، ابتدا به image هاش نیاز داریم. برای نصب کافکا دو ایمیج zookeeper و kafka رو نیاز داریم. به نظرم همواره Image های رسمی خود شرکت ها دانلود بشن بهتر هست. بنابراین برای Zookeeper ایمیج اصلی اون به همین نام که تا لحظه نوشتن این متن 100M دانلود داشته رو استفاده میکنیم. نحوه دانلود همونطور که خودتون بهتر میدونین با استفاده از دستور زیر در cmd انجام میشه.(داکر رو لاگین کرده باشین و vpn روشن بشه) docker pull zookeeperدر زمان نوشتن این متن آخرین نسخه ی zookeeper که در دسترسه ورژن 3.5.6 هست که من برای نصب چون ورژن تعیین نکردم همین ورژن دانلود شده. برای اجرا هم فعلا همه چیز به صورت دیفالت استفاده میکنیم، در پست های بعدی نحوه تغییر این مقادیر دیفالت رو با هم بررسی میکنیم. بالا آوردنش ساده است و نیاز هست zookeeper به صورت یک سرویس در فایل docker-compose تعریف بشه. (در اینجا تنها 1 zookeeper بالا آوردم و فعلا قصد کلاستر کردن رو ندارم و این مبحث در پست های آینده مفصل توضیح داده میشه)در zookeeper پورت دیفالت 2181 هستش که ما هم همین رو فعلا استفاده میکنیم و تغییری در اون نمیدیم. همچنین لازمه یه فولدر رو برای اینکه log هاش (در بخش های دیگه با جزییات اشاره میشه کاربردش چی هست رو هم مقدار میدیم). من در اینجا یه آدرس از سیستمم (D:\Projects\Hazard\hazard-detection\data_log) رو براش ست کردم. در بخش environment هم آدرس سرور zookeeper رو میذاریم، من در اینجا localhost رو ست کردم به آدرس 0.0.0.0. zoo1: image: zookeeper hostname: zoo1 volumes: - D:\Projects\Hazard\hazard-detection\data_log:/data-log ports: - 2181:2181 environment: ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181برای image کافکا از confluentinc/cp-kafka:6.0.1 که در زمان نوشتن این نسخه ورژن آخر بود استفاده شده. این شرکت در زمینه پیشبرد و افزودن سرویس به کافکا خیلی فعاله و در شرکت ما از این Image استفاده میشه.(image کافکا از wurstmeister هم خیلی پر کاربرده) مثل مرحله قبل اول pull میکنیم. kafka چون به zookeeper وابسته است لازمه بعدش بالا بیاد بنابراین depends_on zoo1 رو اضافه میکنیم. اینجا هم از پورت دیفالت که 9092 هست استفاده میکنیم و به بیرون expose اش میکنیم. آدرس zookeeper چون اینها خارج از محیط داخلی داکر با هم پیام رد و بدل میکنن به آدرس IP سیستم ست میکنیم که در اینجا IP سیستمم 192.168.160.93 هستش و در بالا هم دیدیم zookeeper بر روی پورت 2181 بالا می آید و نحوه انتقال پیام ها که به صورت PLAINTEXT باشه رو در بخش environment اش ست میکنیم. همچنین لازمه آدرس سروری که کلاینت ها از طریق اون به کافکا متصل میشن و کافکا روی اون بالا اومده رو در بخش KAFKA_ADVERTISED_LISTENERS قرار میدهیم که من آدرس سیستم خودم رو قرار دادم.kafka1: image: confluentinc/cp-kafka:6.0.1 depends_on: - zoo1 ports: - 9092:9092 environment: - KAFKA_ZOOKEEPER_CONNECT=192.168.160.93:2181 - KAFKA_LISTENERS=PLAINTEXT://:9092 - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.160.93:9092با دستور docker compose up کافکا و zookeeper بالا میان و همدیگه رو میشناسن! کل فایل docker-compose.yml در ادامه آوردم.version: "3.6" services: zoo1: image: zookeeper hostname: zoo1 volumes: - D:\Projects\Hazard\hazard-detection\data_log:/data-log ports: - 2181:2181 environment: ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181 kafka1: image: confluentinc/cp-kafka:6.0.1 depends_on: - zoo1 ports: - 9092:9092 environment: - KAFKA_ZOOKEEPER_CONNECT=192.168.160.93:2181 - KAFKA_LISTENERS=PLAINTEXT://:9092 - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.160.93:9092در این نوشتار، نصب و کانفیگ کافکا به دو روش بیان شد. کافکا بر روی ویندوز تنها از طریق دانلود و extract محتوای آن قابل استفاده است، نیاز به نصب خاصی ندارد. کانفیگ آن نیز تنها با مقدار دهی دو متغیر آن انجام پذیر است و بسیار راحت اجرا می شود. با استفاده از داکر نیز در صورتی که مختصری با داکر آشنایی داشته باشین به راحتی قابل اجرا است. از اینکه وقتتون رو در اختیارم قرار دادید و متن رو خوندید ممنونم، اگر سوالی یا پیشنهادی داشتین خوشحال میشم برام کامنت بذارید.برای مطالب این نوشتار از درس های Getting Started with Apache Kafka و Designing Event-driven Applications Using Apache Kafka Ecosystem از pluralsight استفاده شده است.

تکنیک لیبل زدن به داده ها- بخش دوم: روش Active Learning

مریم محمدی — Wed, 12 Jun 2024 13:23:38 +0330

در بخش قبل، روش لیبل زنی به داده ها با استفاده از روش Semi-supervised learning رو بررسی کردیم. در این بخش، نحوه ی لیبل گذاری با استفاده از Active لرنینگ بیان میشه.در تکنیک لیبل زنی با کمک active learning، سعی بر این است که به نحوی داده ها را برای لیبل خوردن جداسازی کرده به نحوی که در فرآیند train بیشترین اطلاعات را داشته باشیم.در صورتی که تنها تعداد کمی داده ی لیبل شده داریم و یا مثلا expert ای داریم که با بودجه ی پروژه تعداد کمی داده ی انتخاب شده ی ما را بررسی و لیبل گذاری میکند، این روش کمک میکند که به صورت متعادلتری sample برای لیبل زنی و استفاده در آموزش انتخاب کنیم.این روش ادعا میکند در صورت استفاده، متریک هایی مثل accuracy را بهبود میدهد.(جلوتر میبینیم جزییاتش رو ولی خلاصه بگم سعی میکنیم تمامی فیچر space با داده های انتخابی پوشش داده میشه برای همین دقت بالا میرود.)نحوه انجام active learning به طور کلی به صورت زیر است:1- ابتدا از میان داده های بدون لیبل با کمک active learning sampler داده های خاصی را انتخاب میکنیم.2- در مرحله دوم یک annotator مانند انسان یا روش های دیگر این سمپل های خاص لیبل گذاری میشوند.3- در مرحله سوم با این داده های جدید، یک دیتاست از داده های لیبل خورده تشکیل میدهد و در مرحله train مدل نهایی مورد استفاده میشود.4- چرخه ی بالا از مرحله ی 1 تا 3 دائم تکرار میگردد. تا به تعداد داده های مورد نیاز یا دقت مورد نظر برسیم.فرآیند تکنیک Active Learningبخش اصلی فرآیند active learning نحوه sampling داده هاست که روش های متقاوتی دارد:1- Margin Sampling: در این روش لیبل گذاری در داده هایی که بالاترین عدم قطعیت دارند هر بار جدا میشوند.2- Cluster-based sampling: در این روش، یک دسته از داده های مختلف/گوناگون انتخاب میشود! این داده ها از طریق cluster شدن داده ها در فضای feature space انتخاب میشوند.3- Query by committee: در این روش، مدلهای متفاوتی با کمک داد های لیبل خورده آموزش داده میشود. سپس داده های بدون لیبل با کمک این مدلها تست شده و داده ای که بیشترین disagreement و اختلاف در بین لرنر ها برای لیبل گذاری آن وجود دارد، انتخاب میشود.4- Region-based sampling: این روش، جدیدترین روش است. داده ها براساس فیچرها به region های مختلف تقسیم میشوند، سپس با یکی از روش های انتخاب داده ی گفته شده در قبل، لیبل گذاری میشوند.در ادامه تنها به روش Margin sampling با جزییات میپردازیم و جزییات روش Margin Sampling برای بقیه روش ها هم صادقه.Margin sampling: این روش با استفاده از یک classifier بر روی داده هایی که از ابتدا لیبل دارند، یک مدل ساده آموزش میدهد. سپس با استفاده از آن، کم محتمل ترین نمونه بدون لیبل، داده با score نزدیک به 0.5، که نزدیک به فضای جدا کننده ی داده هاست، برای لیبل گذاری به annotator داده میشود.انتخاب داده با کمترین مقدار قطعیتدوباره با کمک داده های جدید مدل مجدد لرن شده و فرآیند margine sampling و active learning تکرار میگردد.آموزش مجدد مدل و انتخاب دومین داده بعد از لیبل زنی داده اولیه با annotatorوقتی داده ها با استفاده از روش active learning انتخاب میشود، accuracy سیستم خیلی بالاتر میرود. در نمودار زیر accuracy سیستمی که داده های آن با active learning برای لیبل گذاری انتخاب شده است در مقایسه با انتخاب رندم داده ها آمده است.میزان دقت مدل در حین learning با کمک داده های انتخاب شده در روش active learningنمودار بالا نشان میده که با تکنیک active learning میتوان با لیبل گذاری تعداد کمتری از داده به accuracy بالاتری رسید.در روش های sampling همیشه داده ای برای لیبل زنی توسط annotator انتخاب میشه که از فضای لیبل خورده در حقیقت دورتره! این به این معنیه که داده ی انتخاب شده، در feature space توسط سایرین پوشش داده نشده که در موردش بیشترین ابهامه، بنابراین با انتخابش و لیبل زنی بر روی اونها میتونیم بیشترین فضای فیچرها رو پوشش داده باشیم و با همین حربه دقت مدل مون بالا میره.نمونه کد active learning با کمک margin sampling در ادامه آمده است.from sklearn.svm import SVC from sklearn.datasets import make_blobs from sklearn.model_selection import train_test_split import numpy as np iris = datasets.load_iris() # Load a dataset (e.g., the Iris dataset) X = iris.data y = iris.target # Split the data into initial labeled and unlabeled sets X_labeled, X_unlabeled, y_labeled, y_unlabeled = train_test_split(X, y, test_size=0.9, random_state=42) clf = SVC(probability=True) # Initialize the SVM classifier clf.fit(X_labeled, y_labeled) # Train the classifier on initial data # Active learning loop for _ in range(10): # Repeat 10 times # Predict probabilities on the unlabeled data y_prob = clf.predict_proba(X_unlabeled) # Calculate the margin for each instance margins = np.abs(y_prob[:, 0] - y_prob[:, 1]) query_idx = np.argmin(margins) # Select the smallest margin # Label the selected instance: X_labeled = np.vstack([X_labeled, X_unlabeled[query_idx]]) y_labeled = np.hstack([y_labeled, y_unlabeled[query_idx]]) X_unlabeled = np.delete(X_unlabeled, query_idx, axis=0) y_unlabeled = np.delete(y_unlabeled, query_idx) clf.fit(X_labeled, y_labeled) # Retrain the classifier accuracy = clf.score(X, y) # Evaluate the final classifier print("Final classifier accuracy:", accuracy)در این مطلب روش active learning برای لیبل زنی به داده های بدون لیبل رو دیدم. در بخش سوم نحوه لیبل زنی روش Weak supervision رو هم با هم میبینیم. از اینکه وقت خود را در اختیار من قرار دادین بسیار سپاس گزارم. امیدوارم این مطلب برای شما موثر و مفید باشد.

تکنیک لیبل زدن به داده ها- بخش اول: روش Semi supervised

مریم محمدی — Wed, 12 Jun 2024 13:12:58 +0330

سلام، حدود یک سال میشه که شرکت ما به سمت پروژه های هوش مصنوعی رفته و من جز تیمشونم. طی یه سری پست های کوچیک میخوام مطالبی که خودم یاد گرفتم و برام جالبه به عنوان فردی که فقط مفاهیم پایه رو میدونسته و هوش مصنوعی رو صرفا تئوری بلد بوده با شما به اشتراک بذارم. به این سری پست ها، هوش مصنوعی در عمل میگم. مطالب رو از سایت coursera و سری درسهای Machine Learning In Production که توسط DeepLearning.AI ارائه شد، یاد گرفتم که به همه پیشنهاد میکنم درسهاشو بگذرونن(درس 3 و 4 کاملا مرد افکنه! )روش های یادگیری خودتون بهتر میدونین که میتونه به صورت supervised هم باشه که در این صورت لازم داریم داده هامون لیبل خورده باشن. خب داده ی لیبل خورده خیلی راحت پیدا نمیشه، لیبل زدن پر هزینه است ولی داده ی بدون لیبل فت و فراوونه(بعضی وقتا اونم نیست!)در ادامه روش هایی که میتونیم با هزینه ی کم و به روش های کاربردی به داده ها لیبل بزنیم رو بیان کنم. به صورت کلی سه روش زیر برای لیبل زدن وجود داره:Semi-supervised learningActive learningWeak supervision with snorkelدر ادامه ی این روش ها در سه پست کوتاه اومده.Semi-supervised learningدر روش semi-supervised با استفاده از تعداد کمی داده های لیبل دار در ترکیب با تعداد زیادی از داده های بدون لیبل، به نحوی لیبل ها را استنتاج میکنیم. برای مثال میتوانیم این داده ها در فضای فیچرها کلاستر کنیم و تو هر دسته لیبل موجود رو به بقیه هم نسبت میدیم. روش های دیگه ای هم هست مثل label-propagation یا graph-based ها که تکنیک های دیگه ی این روش لیبل گذاری هستن. همه ی این روش ها در حقیقت transudative learning هستن که یجوری میخوان این لیبل ها رو منتقل کنن! شاید باورتون نشه ولی این ترکیب کردن داده های لیبل گذاری شده و بدون لیبل accuracy سیستم رو زیاد میکنه!(خودم تست گرفتم واقعا زیاد شد!!!)Label Propagationلیبل propagation یکی از الگوریتم های این دسته است. براساس community structure لیبل میزنه. بیشتر در داده های گراف بیس کاربردیه مثل detect کردن community در داده ها یا آنالیز network ها. با استفاده از پترن لینک های موجود لیبل را به باقی هم propagate میکند. این روش همچنین در تشخیص آنومالی، segmentation تصاویر، کاهش بعد که داده هاشون گراف بیس نیست هم کاربرد داره. نمونه پیاده سازی این روش با استفاده از scikit-learn به صورت زیر هستش:from sklearn import datasets from sklearn.semi_supervised import LabelPropagation from sklearn.metrics import accuracy_score iris = datasets.load_iris() # Load a sample dataset (e.g., the Iris dataset) X = iris.data y = iris.target # Create a mask for the labeled data points (only the first 10 data points are labeled) labeled_mask = [True] * 10 + [False] * (len(X) - 10) label_prop_model = LabelPropagation() # Initialize the Label Propagation model label_prop_model.fit(X, y, labeled_mask) # Fit the model using the labeled and unlabeled data predicted_labels = label_prop_model.transduction_ # Predict the labels for all data pointsلیبل propagation تکنیک های مختلفی داره و حتی میشه براساس کلاسترینگ هم اینکار رو انجام داد که ازش میگذریم.در این مطلب روش semi-supervised برای لیبل زنی به داده های بدون لیبل رو دیدم. در بخش دوم و سوم نحوه لیبل زنی روش های Active learning و Weak supervision رو هم با هم میبینیم. از اینکه وقت خود را در اختیار من قرار دادین بسیار سپاس گزارم. امیدوارم این مطلب برای شما موثر و مفید باشد.

گرافانا و پرومتئوس-بخش اول

مریم محمدی — Sun, 27 Aug 2023 13:50:14 +0330

تازگیا تو شرکت لازم داشتیم سیستم مانیتورینگ بنویسیم. کار ما مرتبط به تراکنش های بانکی هستش، در کنار اینکه چقدر RAM، CPU و ... استفاده شده، باید یکم از اطلاعات بیزینسی مثلا چند تا تراکنش انجام شده، تا الان کدوم مراحل از روند انجام شده و ... رو نشون بدیم. وقتمون هم خیلی کم بود. به همین خاطر دنبال سریع ترین و کم خطاترین راه پیاده سازی (در عین حال راحتترین) بودیم که متوجه شدیم Prometheus چقدر خوبه!!!! برای UI سیستممون از Grafana استفاده کردیم که هیچ زحمتی برای UI نداشتیم! هر دو Prometheus و Grafana خیلی ساده هستن. Prometheus اطلاعات مورد نیاز برای نشون دادن رو ایجاد میکنه و Grafana اونها رو نمایش میده. برای همین اول Prometheus رو میگم که چی هست و چطور کار میکنه، سپس Grafana و در نهایت نصب و دمویی رو در غالب چند پست میگم. در این نوشته صرفا نحوه کار پرومتئوس بیان شده. در قسمت دوم اینکه چطوری پرومتئوس و گرافانا رو بالا بیاریم و یک سمپل کوچیک از نحوه ی دریافت اطلاعات رو میبینیم. در بخش سوم متریک های موجود و در قسمت آخر نحوه ی کار گرافانا رو با هم میبینیم.پرومتئوس چیستپرومتئوس open source، free و toolkit مناسب مانیتورینگ است. بر روی لینوکس و ویندوز هر دو یکسان کار میکند و cross platform است. از روی نرم افزارها یا سیستم هایی که میخوایم اونها رو مانیتور کنیم یه سری اطلاعات به دست میاره که بهشون متریک میگه. به خود این نرم افزار ها که قصد به دست آوردن اطلاعات ازش رو داره هم تارگت میگه. این متریک ها شامل میزان CPU، memory، میزان استفاده از دیسک و ... یا متریک های custom که کاربر آن را تعریف میکنه، است. پرومتئوس متریک ها را به صورت polling در بازه های زمانی مشخص از آدرس تارگت هایی که بهش معرفی کردیم، میگیره و در دیتابیس time series خودش نگهداری میکنه. بنابراین میتوان با query زدن روی یک range زمانی، داده ها را به دست آورد که برای اینکار نیز یک API نوشته که به راحتی بشه با اون query زد. همینطور Prometheus یک وب اپ ساده هم برای کارهای administrator و query زدن دارد. یک سیستم alerting نیز دارد که میتوان با کمک آن، در صورت بروز مشکلاتی به کاربر با email یا message پیغام داد.با پرومتئوس همزمان میتوان سیستم ها را جداگانه ولی بخشی از یک کل، مانیتور کرد. نحوه ی به دست آوردن متریک ها در تمامی سیستم ها فارغ از اینکه سرور، وب اپ و ... باشد، با یکدیگر یکسان است. متریک ها همونطور که گفتیم یا خودمون ایجاد کردیم یا از ابزارهای آماده ای که به اونها exporter گفته میشود برای ما به دست می آید. در تمامی تارگت ها، خود تارگت به روشی که در آینده میفهمیم، این متریک ها را در آدرسی قرار میده و پرومتئوس آنها را میخواند که به این روش scrape کردن میگویند. در هر دو روش، در یک آدرس مشخص این اطلاعات را قرار میگیرد و پرومتئوس با روش polling ودر فاصله زمانی مشخصی که میتوان schedule کرد، آنها را میخواند.پرومتئوس متریک هاشو با فرمت خاصی نشان میده. این فرمت خیلی ساده است. به این صورت که اسم متریک و مقدارش هستش. در کنار این فرمت و مقدار، metadata هایی شامل description یا تایپ داده هم وجود داره که توسط خود پرومتئوس گذاشته میشه.به علاوه هر داده شامل یک سری label هم میتونه باشه که بعدا در مورد label توضیح میدم. در شکل زیر دو متریک به همراه metadata های اون اومده. متریک اول node_filefd_allocated هست و دومی node_disk_io_time_seconds_total که این متریک لیبل device هم داره. در شکل دو خط کامنت قبل از مقدار هر متریک اومده،خط ابتدایی description ای که تعریف متریک است و دیگری برای تایپ که در اینجا gauge است. متریک ها 4 دسته هستند: gauge، counter، histogram و summary که تعریف دقیقشون رو در قسمت های بعد میبینیم.نمونه ای از فرمت پرومتئوس

الگوهای طراحی نرم افزار-پترن های رفتاری-Iterator and mediator Pattern

مریم محمدی — Tue, 19 Jan 2021 08:55:00 +0330

پترن Iteratorزمانی که بخواهیم بر روی یک مجموعه از آیتم ها پیمایش انجام دهیم، میتونیم از پترن iterator استفاده کنیم.در صورتی که بخواهیم وظیفه‌ای پیمایشی آیتم‌های مجموعه رو همزمان بر عهده مجموعه قرار بدیم کار پیچیده میشود، به هم دلیل، کار پیمایش را از آن مجموعه جدا می کنیم. این آبجکت ها که وظیفه پیمایش بر روی مجموعه را دارند، iterator می گوییم. ما به خود مجموعه میگیم که فلان iterator را به ما بده و ما با استفاده از آن، بر روی مجموعه پیمایش انجام میدهیم.این آبجکت iterator لازم است به اعضای مجموعه دید داشته باشد و بنابراین encapsulation بین آبجکت های داخل مجموعه از بین رفته است ولی در عوض client هایی که میخواهند از آن استفاده کنند، دیگر این encapsulation را نقض نمیکنند. نحوه پیمایش با کمک iterator به صورت های مختلف بسته به نوع iterator میتونه متفاوت باشه برای مثال با الگوریتم bfs یا dfs پیاده سازی شده است.تمام این مدل های مختلف iterator اینترفیس ثابت را پیاده سازی می کند یعنی در هر زبانی اگر بگوییم یک iterator میخواهم، اینترفیس ثابتی به ما برمیگردد که شامل next، first یا hasNext (ممکنه isDone هم بگن) و currentItem هستش که از اثرات این پترن است که در تمامی زبان ها مفهوم یکسان پیاده سازی شده است. بنابراین وظیفه پیمایش بر روی مجموعه‌ها از اون بیرون اومده و این وظیفه به خود iteratorها محول شده و در list بیان می‌کنم که فلان iterator رو به من بده.این iterator ها میتوانند همان زمانی که currentItem را صدا میکنیم آبجکت جدید را بسازند و لزومی ندارد که صرفا بر روی یک لیست یا collection با داده های از قبل آماده باشند.پترن Mediatorوقتی تعداد زیادی آبجکت داریم، که اینها به طرز پیچیده ای با هم کار میکنند. حالا برای از بین بردن coupling بین بخش های مختلف وسط اینها یک مدیر قرار میدیم که همه با این ارتباط در ارتباط باشند. به این مدیر یک mediator یا میانجی میگن. اینطوری coupling بسیار کم میشه ولی دیگر تمامی بخش های مختلف به این couple میشوند که این آبجکت میانجی میتونه تبدیل به God class بشه و بسیار خطرناکه و مگه خیلی بی ارزه که از این پترن استفاده کنیم چون باید همیشه مراقب آن باشیم. برای مثال یکی از مواردی که خیلی این پترن در آن کاربرد داره این هستش که سیستم کاملاً stable شده ولی ماژولهای آن خیلی در هم تنیده شده اند و نیاز به maintenance و تغییرات خاصی در آینده نداشته باشه. بنابراین سیستم به نحوی stable شده اما حالا بخواهیم coupling که بین بخش ها رو کم کنیم، میتوانیم یک میانجی یا مدیاتور این بین قرار بدیم.بنابراین وقتی کار ارتباط بخش های مختلف را از آبجکت ها جدا میکنیم آیجکت ها خودشان ساده میشن.بنابراین فهمیدن کد آبجکتها بسیار ساده تر میشه. ولی ایرادی که داره این هستش که این میانجی خودش bottle neck میشه چون همه می خوان فقط در این ارتباط با این آبجکت باشن.

راه اندازی لینوکس با استفاده از windows subsystem

مریم محمدی — Sun, 03 Jan 2021 10:25:40 +0330

من به دلیل محدودیت های شرکتم نمیتونم لینوکس رو مستقیم بر روی سیستمم نصب کنم، به همین خاطر دنبال راهی بودم که این قوانین نقض نشه و من هم لینوکس رو داشته باشم! راهی پیدا کردم با استفاده از windows subsystem بود. این راه اندازی به نظرم جالب بود بنابراین در ادامه ابتدا نصب لینوکس با کمک فیچرهای ویندوز 10 را بیان میکنم.در ویندوز، subsystem ای را فراهم آورده که بتوان لینوکس را بر روی آن نصب کرد. این ویژگی در حالت عادی فعال نیست و برای فعال کردن آن لازم است ابتدا Turn Windows Features on or off را در منوی start جستجو کرده سپس دو ویژگی ویندوز که نیاز به فعال شدن دارد و در شکل (1) نشان داده شده را فعال نمایید. سپس با کمک Microsoft Store ابونتو را نصب کردم و ابونتو به عنوان یک اپلیکیشن بر روی سیستمم نصب شد. به همین راحتی لینوکس واقعی ( و نه virtual) را بر روی سیستم ویندوزیم نصب کردم. بر روی این لینوکس به علت محدودیت های subsystem، نمیشه gnome نصب کرد و از terminal صرفا میشه ازش استفاده کرد و برای من جالب و کار راه بندازه.شکل(1) ویژگی های مورد نیاز برای فعال کردن windows subsystemشکل(2) نصب پوسته ی لینوکس با استفاده از Microsoft Store

الگوهای طراحی نرم افزار-پترن های ساختاری-Facade Pattern

مریم محمدی — Tue, 13 Oct 2020 16:52:53 +0330

من دوست دارم مطالبی که یاد میگیرم را با افراد دیگر نیز مطرح کنم. در این نوشتار به توضیح پترن façade از سری پترن های structural از GoF پرداخته میشود.پترن façade برای ساده تر شدن استفاده از APIها به کار میرود. در صورتی که کدها از ساختار مناسبی برخوردار نباشند و یا استفاده از آنها برای کاربر گنگ باشد میتوان با استفاده از این پترن، استفاده از API را ساده تر کرده و یا مواردی که لازم است توسط کاربر دیده نشود، از او بپوشانیم.این پترن یک پترن refactoring است. یعنی در ابتدا کد نامناسب یا پیچیده ای نوشته شده است و میخواهیم با کمک این پترن استفاده از کد را بهبود ببخشیم. کلمه ی façade به معنای نمای خارجی و منظر( منظر به معنی دریچه ی دیدن) است. کارکرد پترن Façade نیز از معنای نام آن پیروی میکند یعنی با استفاده از این پترن دریچه ی دیدی به یک subsystem ایجاد میشود. در استفاده های امروزی از façade تنها ارتباط به subsystem از طریق آبجکت façade است و در استفاده های قدیم تر از این پترن امکان استفاده ی مستقیم از subsystem در کنار façade object وجود داشت که اکنون این مورد در پترن façade مورد استفاده نمیباشد و تنها از طریق façade object ممکن است. شکل 1 نحوه ی کارکرد پترن فساد را نمایش میدهد.شکل(1) نحوه کارکرد façade patternپترن façade یک interface واحد بر روی کل subsystem ایجاد میکند. بدون این پترن، استفاده کننده از subsystem لازم بود که interface هر یک از بخش های را بشناسد، ترتیب استفاده از آنها و پیش نیاز های مورد نیاز را برای استفاده بداند. بنابراین کلاس های بیرونی subsystem دیگر با تمامی کلاس های داخلی در ارتباط نیست و کار Instance گرفتن، سرویس گرفتن و ... را خودش انجام میدهد. در حقیقت کار چرخوانی انجام میدهد. این کلاس قرار نیست هیچ گونه پردازشی انجام دهد بلکه کارها بر عهده ی کلاس های داخلی subsystem است و این کلاس آن ها را فراخوانی میکند. مزایای استفاده از پترن façade در شکل 2 آمده است.شکل(2) مزایای استفاده از پترن façadeدر نامگذاری کلاس façade لازم است کلمه ی façade ذکر شود تا تمامی تیم برنامه نویسان بدانند که این کلاس یک Façade است زیرا به محض انجام پردازشی توسط خودش، کلاس به حالت God class در می آید بنابراین در نامگذاری دقت شود تا در زمان maintenance پردازشی انجام ندهد و برنامه نویسان این کلاس را سنگین نکنند.این الگوریتم coupling را بسیار پایین می آورد زیرا ما در ارتباط با تنها کلاس façade هستیم ولی generality ندارد زیرا تنها محدود به اینترفیس آن میتوان از آن استفاده کرد در حالی که اگر از خود subsystem مستقیم استفاده میشد ممکن بود به تنوع بیشتری دست یافت. این روش مزیت سادگی در استفاده را برای کلاینت به همراه دارد. میتوان برای کلاینتهای مختلف چندین Façade متفاوت بر روی یک subsystem تعریف کرد تا بتوان بحث generality را نیز گسترش داد. بنابراین subsystem محدود به تنها یک façade نیست.الگوی façade با کمک composition کار میکند. بنابراین در صورتی که نیاز به استفاده از inheritance بود، در آنجا کاربرد ندارد.موارد استفاده شده از این پترن در جاوا به صورت زیر است:java.net.URLjavax.faces.context.FacesContextیک مثال کاربردی از آن در شکل 3 آمده است. در اینجا تنها با صدا کردن کلاس compiler که در اینجا façade class است، دیگر نیازی به نمونه گرفتن از نیازمندی های کامپایل کردن مانند tokenizer، parser مبرا شده و تنها یک instance از compiler و فراخوانی متد compiler کفایت میکند. شکل(3) نمونه ای استفاده از پترن façadeنمونه ی دیگری از مثال استفاده از پترن façade در شکل 4 آمده است. در این مثال کاربر با کمک کلاس HomeTheaterFacade به راحتی میتواند با سیستم صوتی تصویری خود ارتباط برقرار کند و نیازی به دانستن نحوه ی استفاده ی جداگانه از هر کدام از کلاس ها نیست و استفاده از سیستم راحتتر شده است.شکل(4) نمونه ای از استفاده از پترن façadeموارد مورد استفاده از façade به صورت زیر است.1- اغلب زمانی که درخواست میشود سیستم wrap شود و روی آن یک interface تعریف شود منظور داشتن façade است که به داخل subsystem ارتباطی برقرار نباشد. مکان هایی که خیلی stable نیستند و دائم تغییر میکنند برای جلوگیری از انتشار این تغییر به جاهایی که به این بخش ها مرتبط هستند، روی آنها façade تعریف میشود. بنابراین می توان بخش های غیر stable را پوشاند و بقیه را از این تغییرات محفوظ نگه دارند.2- کلاس های façade نماینده ی subsystem است به همین علت میتوان تغییرات کل subsystem از طریق این نماینده به subsystem های دیگر اطلاع داده شود. بنابراین observer بودن از طریق façade ها اعمال شود.3- در معماری لایه ای برای اینکه ارتباطی با داخل لایه برقرار نباشد در هر لایه یک façade قرار داده میشود. تا لایه ها از داخل هم به هم مرتبط نباشند و لایه ها به صورت یک component باشند. زمانی که برنامه نویسی مبتنی بر component معرفی گردید، استفاده از پترن façade گسترش پیدا کرد.دوستی داشتم که میگفت فساد کنین که باعث پیشرفت میشود! شما هم فساد کنین کدهاتون قطعا پیشرفت میکند.از اینکه زمان گذاشتید و مطالعه کردید سپاس گزارم. امیدوارم این مطلب برای شما مفید بوده باشد.منابع مورد استفاده در این نوشتار:کلاس Design Patterns in Java: Structural By Bryan Hansen از سایت Pluralsight، کلاس درس طراحی الگو استاد رامسین در دانشگاه شریف. کتاب head first design pattern.