نقل الـ Events
الـ event = record صغير immutable فيه تفاصيل حاجة حصلت ووقتها. بيتكتب مرة من producer (publisher) ويتقرا من consumers (subscribers)، والـ events المرتبطة بتتجمع في topic أو stream.
ليه مش polling على داتابيز؟ كل ما تقلل الفترة، الـ overhead بيزيد — الأحسن إن الـ consumer يتبلّغ أول ما الحدث يحصل.
Messaging Systems — سؤالين مهمين
- لو الـ producer أسرع من الـ consumer؟ إما ترمي رسايل، أو تعمل buffer في queue (والـ queue لو كبر هيروح فين؟)، أو backpressure (توقف الـ producer — زي Unix pipes و TCP).
- لو نود وقعت، الرسايل تضيع؟ الـ durability ليها تمن (كتابة ديسك أو replication). أحيانًا ضياع شوية مقبول (sensor readings) — بس في عداد مثلًا، أي رسالة ضايعة = رقم غلط.
- Direct messaging (من غير وسيط): UDP multicast (البورصات)، ZeroMQ، StatsD، webhooks. سريع بس محدود في الـ fault tolerance.
- Message brokers (queues): سيرفر في النص، بيستحمل ان الكلاينتس تيجي وتمشي، والـ consumers غالبًا async.
كل رسالة لـ consumer واحد (توزيع الشغل). (JMS: shared subscription)
كل رسالة لـ كل الـ consumers (topic subscriptions).
والـ Acknowledgment: الـ consumer يأكد إنه خلص، وإلا الـ broker يعيد إرسالها لحد تاني. بس مع load balancing + redelivery → الترتيب بيتلخبط. لو الترتيب مهم: queue منفصل لكل consumer.
Partitioned Logs (Kafka) ⭐
الـ brokers التقليدية (AMQP/JMS) بتمسح الرسالة بعد الـ ack — مش زي الداتابيز. يعني مينفعش ترجع تقرا القديم، وconsumer جديد يشوف الجديد بس. الفكرة: نجمع durability الداتابيز مع low-latency الـ messaging → Log:
أمثلة: Apache Kafka، Amazon Kinesis، Twitter DistributedLog. (ملايين الرسايل في الثانية بالـ partitioning، والـ replication للـ fault tolerance.)
| AMQP/JMS (RabbitMQ…) | Log-based (Kafka) | |
|---|---|---|
| بعد المعالجة | الرسالة تتمسح | تفضل (retention بالحجم أو المدة — ring buffer على الديسك، ممكن أيام أو أسابيع) |
| التوزيع | رسالة رسالة لأي consumer | partition كامل لـ consumer واحد (عدد الـ consumers ≤ عدد الـ partitions) |
| الترتيب | ممكن يتلخبط | مضمون جوه الـ partition |
| التتبع | ack لكل رسالة | offset واحد (زي log sequence number في الـ replication) |
| القراءة من الأول | ❌ | ✅ Replay — زي الـ batch: القراءة مش بتغير حاجة |
| مناسب لـ | Task queues، المعالجة مكلفة والترتيب مش مهم | Throughput عالي، معالجة سريعة، والترتيب مهم |
الداتابيز والـ Streams
الـ replication log في الفصل 5 أصلًا stream of events! والفكرة هنا: نستخدم الـ streams عشان نخلي كذا سيستم متزامنين (داتابيز + cache + search index + data warehouse).
Change Data Capture (CDC)
بدل dual writes: خلي الداتابيز هي الـ leader، واقرا الـ replication log بتاعها كـ stream وطبّقه على الأنظمة التانية بنفس الترتيب. أمثلة: LinkedIn Databus، Facebook Wormhole، Debezium و Maxwell (MySQL binlog)، Bottled Water (Postgres WAL)، Mongoriver. وغالبًا async.
- Initial snapshot: مش هتحتفظ بالـ log كله للأبد — تاخد snapshot وتعرف الـ offset بتاعه.
- Log compaction (Kafka): يحتفظ بآخر قيمة لكل key بس (والـ delete = tombstone). فالـ topic يبقى فيه نسخة كاملة من الداتابيز تقدر تبني منها أي derived system من الصفر.
- وداتابيزات بقت بتدعم change streams كـ feature (RethinkDB, Firebase, CouchDB, Meteor, VoltDB، و Kafka Connect).
Event Sourcing
فكرة من مجتمع DDD: خزّن كل التغييرات كـ events immutable على مستوى الأبلكيشن (مش على مستوى الداتابيز زي CDC).
low-level: الداتابيز بتتعدل عادي (mutable)، والـ log بيتطلع منها. الأبلكيشن مش واخد باله.
الأبلكيشن نفسه بيكتب events بتعبر عن نية اليوزر (“طالب لغى تسجيله في كورس”) مش “اتمسح صف”. الـ log هو مصدر الحقيقة.
- الـ state الحالي = تطبيق كل الـ events من الأول (مع snapshots للتسريع). ومينفعش compaction لأن الـ event الأحدث مش بيلغي اللي قبله بالضرورة.
- Commands vs Events: الطلب الأول command ممكن يترفض (validation). لما يتقبل يبقى event = حقيقة (fact) مش ممكن تتغير. لو اليوزر لغى بعدين = event جديد.
مميزات الـ Immutability
الحدود: داتا بتتغير كتير → التاريخ يبقى ضخم والـ compaction/GC مهم. وأحيانًا لازم تمسح فعلًا لأسباب قانونية/خصوصية (excision في Datomic) — وده صعب لأن النسخ في كل حتة (backups وغيرها).
معالجة الـ Streams
تعمل إيه بالـ stream؟ (1) تكتبه في داتابيز/كاش/index، (2) تبعته للمستخدمين (إيميل، notification، dashboard)، (3) تعالجه وتطلع stream جديد (pipeline من operators زي Unix). والفرق الأساسي عن الـ batch: الـ stream مش بيخلص أبدًا — فمفيش sort-merge joins، والـ fault tolerance مختلف (مش هتعيد تشغيل job شغال سنين من الأول).
| الاستخدام | الفكرة |
|---|---|
| Complex Event Processing (CEP) | تخزن الكويري وتعدّي عليها الـ events (عكس الداتابيز!). بتدور على patterns معينة (Esper, IBM InfoSphere). |
| Stream analytics | aggregations ومقاييس على windows: معدل حدث، متوسط متحرك، مقارنة بالفترة اللي فاتت. وفيه probabilistic algorithms (Bloom filter، HyperLogLog). (Storm, Spark Streaming, Flink, Samza, Kafka Streams, Google Dataflow). |
| Materialized views | تحديث cache/search index/view من الـ changelog باستمرار — محتاج كل الأحداث من الأول مش window بس. |
| Search on streams | تخزن queries وتطابق عليها المستندات الجاية (Elasticsearch percolator) — تنبيهات عقارات مثلًا. |
| Actors vs Streams | الـ actors مش مخصصة للـ data management، والرسايل غالبًا ephemeral وممكن تلف دواير. أما الـ stream processing فـ pipelines acyclic وdurable. |
التعامل مع الوقت ⏰
الوقت اللي الحدث حصل فعلًا (مسجّل في الـ event).
الوقت اللي الـ processor شاف فيه الحدث (ساعة الماكينة).
- Stragglers: إمتى تقول الـ window خلص؟ أحداث متأخرة ممكن توصل بعد ما تقفل. الحل: تتجاهلها (وتراقب عددها) أو تبعت تصحيح للنتيجة.
- ساعة الموبايل مش موثوقة: سجّل 3 timestamps: وقت الحدث على ساعة الجهاز، وقت الإرسال على ساعة الجهاز، وقت الاستلام على ساعة السيرفر. الفرق بين التانيين = offset الساعة → صحح الأول.
أنواع الـ Windows
ثابت ومن غير تداخل: كل دقيقة لوحدها (10:03:00–10:03:59).
ثابت ومتداخل: 5 دقايق كل دقيقة (10:03–10:08، 10:04–10:09).
كل الأحداث اللي بينها أقل من مدة معينة (buffer مترتب بالوقت).
مالوش مدة ثابتة: كل أحداث يوزر لحد ما يبقى inactive فترة (30 دقيقة مثلًا) — website analytics.
Stream Joins
| النوع | مثال | إزاي |
|---|---|---|
| Stream-stream (window join) | بحث + click على نتيجة: احسب الـ click-through rate. | الـ processor يحفظ state (الأحداث في آخر ساعة مثلًا متفهرسة بالـ session ID)، ولما الحدث التاني ييجي يطابق. ولو مفيش click → event “مفيش”. |
| Stream-table (enrichment) | activity events + profiles: ضيف معلومات اليوزر للحدث. | بدل query لداتابيز بعيدة لكل event، خد نسخة محلية من الجدول وحدّثها بـ CDC (زي broadcast hash join بس الجدول بيتحدث). |
| Table-table (materialized view) | Twitter timeline cache: tweets + follows. | الاتنين changelogs، وكل تغيير في جهة يتعمله join مع آخر state للجهة التانية — زي materialized view لـ join query. |
Fault Tolerance و Exactly-Once
في الـ batch سهلة (task فشلت → أعدها، والـ output يظهر بس لو نجحت). في الـ stream؟ مش هتستنى الـ stream يخلص (مش هيخلص!). الهدف: exactly-once semantics (الأدق: effectively-once) = النتيجة كأن كل حدث اتعالج مرة واحدة حتى لو اتعاد.
| الطريقة | الفكرة |
|---|---|
| Microbatching | Spark Streaming: قسم الـ stream لـ batches صغيرة (~ثانية) وعاملها كـ batch عادي. (حجم الـ batch بيأثر على الـ latency والـ overhead.) |
| Checkpointing | Flink: snapshots دورية للـ state بـ barriers في الـ stream. لو وقع يرجع لآخر checkpoint. |
| Atomic commit | كل الـ outputs والـ side effects (رسايل، كتابات، offset) تحصل مع بعض أو لأ — جوه الفريمورك (Google Cloud Dataflow, VoltDB, Kafka مخطط). |
| Idempotence ⭐ | عملية لو اتعملت مرتين نفس نتيجة مرة. ولو مش idempotent بطبيعتها: خزّن الـ offset مع الكتابة وماتطبقش نفس الـ offset تاني. (محتاج ترتيب ثابت، deterministic processing، و fencing لو نود قديمة لسه حية.) |
| Rebuilding state | الـ state المحلي (counters، join tables) يتخزن remotely أو local مع replication، أو يتعاد بناؤه من الـ input stream/changelog (Kafka Streams، Samza). |
🧪 Lab
شغّل Kafka بـ Docker (image apache/kafka — راجع الـ docs) وجرّب:
- اعمل topic بـ 3 partitions، وابعت رسايل بـ key (user ID).
- شغّل consumer-ين في نفس الـ group — كل واحد هياخد partitions مختلفة.
- اقفل الـ consumers، ابعت رسايل، وارجع: هيكمّلوا من الـ offset.
- اعمل consumer group جديد بـ
--from-beginning— replay لكل حاجة!
kafka-topics.sh --create --topic jobs --partitions 3 --bootstrap-server localhost:9092kafka-console-producer.sh --topic jobs --property parse.key=true --property key.separator=: --bootstrap-server localhost:9092kafka-console-consumer.sh --topic jobs --group g1 --bootstrap-server localhost:9092بونص: جرّب Debezium عشان تعمل CDC من Postgres لـ Kafka.