مقدمة الجزء التالت: Systems of Record vs Derived Data
المكان اللي الداتا بتتكتب فيه أول مرة. لو فيه تعارض، هو الصح. غالبًا normalized.
نتيجة تحويل داتا تانية: cache، index، materialized view، توصيات. لو ضاعت تقدر تعيد بناءها. غالبًا denormalized (مكرر) — عشان سرعة القراءة.
الفرق ده مش في الأداة نفسها، في طريقة استخدامها في الأبلكيشن. والجزء ده كله عن إزاي تبني derived data من سيستم لسيستم.
| النوع | بياخد | المقياس |
|---|---|---|
| Services (online) | ريكوست → رد بسرعة | Response time + availability |
| Batch (offline) | داتا كبيرة ثابتة → output، بيشتغل دوريًا | Throughput |
| Stream (near-real-time) | events أول بأول | Latency أقل من الـ batch |
Batch بأدوات Unix
مثال: عايز أكتر 5 صفحات اتزارت من web server log؟ سطر واحد:
cat access.log | awk '{print $7}' | sort | uniq -c | sort -r -n | head -n 5بيعالج GBs في ثواني. ولو كتبته بكود (hash table في الرام) هيشتغل كويس لو الـ URLs المميزة تدخل الرام. بس sort بيعمل spill للديسك و merge (زي SSTables!) وبيستخدم كل الـ cores — فبيتعامل مع داتا أكبر من الرام بسهولة.
فلسفة Unix (1978) — اللي بقت أساس كل حاجة
- Uniform interface: كل حاجة file descriptor (sequence of bytes)، ومعظم الأدوات بتتعامل مع سطور نص. فأي أداة تتوصل بأي أداة.
- فصل المنطق عن التوصيل: stdin/stdout — البرنامج مش عارف جاي منين ورايح فين (loose coupling — شبه dependency injection / inversion of control).
- Transparency: المدخلات immutable، تقدر توقف الـ pipeline في أي حتة وتعمل
lessوتشوف، وتعيد التشغيل براحتك. - ❌ العيب الكبير: ماكينة واحدة بس. وهنا ييجي Hadoop.
MapReduce و الـ Distributed Filesystems
MapReduce = زي Unix tool بس موزع على آلاف الماكينات. بياخد input ويطلع output من غير side effects. وبدل stdin/stdout، بيقرا ويكتب في HDFS (Hadoop Distributed File System — نسخة open source من GFS بتاعة Google).
إزاي الـ Job بيشتغل
- انت بتكتب الـ mapper والـ reducer بس؛ الفريمورك يعمل الباقي.
- “Put the computation near the data”: الـ mapper بيشتغل على الماكينة اللي عليها الـ block — بيوفّر نقل على الشبكة.
- الـ reducers بيحمّلوا الملفات المترتبة من كل mapper ويعملوا merge (= shuffle).
- Workflows: MapReduce job واحد محدود، فبتربط jobs ورا بعض (output واحد = input التاني) بـ schedulers زي Oozie, Azkaban, Luigi, Airflow. وأدوات أعلى زي Pig, Hive, Cascading, Crunch بتعمل كده أوتوماتيك. (في LinkedIn workflows من 50–100 job عادي.)
الـ Joins في Batch
في الـ batch، الـ join معناه حل كل الـ associations في الداتا (مش lookup لصف واحد). مثال: عندك activity events (يوزر عمل إيه) وعايز تـ join مع جدول profiles (سن اليوزر) عشان تعرف أنهي صفحات مشهورة عند أنهي سن. Query لكل event على الشبكة لداتابيز بعيدة = بطيء جدًا ونتيجة مش deterministic. الحل: خد نسخة من الداتابيز (ETL) وحطها في HDFS جنب الـ events.
| النوع | الفكرة | إمتى |
|---|---|---|
| Sort-merge join (reduce-side) | mapper لكل input يطلع join key (user ID). الـ shuffle بيجيب كل الـ records بنفس الـ key لنفس الـ reducer، ومترتبة (secondary sort: البروفايل الأول وبعده الـ events). الـ reducer يعمل join في الرام بسهولة. | عام — مش محتاج افتراضات. بس تقيل (sort + نقل). |
| Broadcast hash join (map-side) | input صغير يدخل الرام كله في كل mapper كـ hash table، وكل mapper يمسح الكبير. | واحد من الاتنين صغير. |
| Partitioned hash join (map-side) | الاتنين مقسومين بنفس الطريقة (نفس الـ key، hash، وعدد partitions) → كل mapper يحمّل partition واحد من الصغير. (Hive: bucketed map joins.) | الاتنين partitioned بنفس الشكل. |
| Map-side merge join | الاتنين partitioned ومترتبين بنفس الطريقة → merge مباشر. | غالبًا لو output job سابق. |
الـ Output بتاع الـ Batch بيروح فين؟
- Search indexes: Google عملت MapReduce أصلًا عشان تبني index البحث. mappers يقسموا المستندات، reducers يبنوا index لكل partition ويكتبوه في الـ DFS. immutable وسهل يتعاد بناؤه.
- Key-value stores (machine learning: classifiers، توصيات): ماتكتبش record record في داتابيز production من جوه الـ mapper! (بطيء، هيضغط على الداتابيز، وبيكسر الـ all-or-nothing لو الـ job فشل). الصح: الـ job يبني ملفات داتابيز جديدة كاملة (Voldemort, Terrapin, ElephantDB, HBase bulk load) والسيرفرات تبدّل عليها atomically.
Hadoop vs MPP Databases
| MPP (Teradata، …) | Hadoop | |
|---|---|---|
| التخزين | داتا لازم تتعمل لها model مقدمًا (schema). | أي bytes — “dump it in” = data lake. الـ schema وقت القراءة (sushi principle: raw data is better). |
| المعالجة | SQL بس — مُحسّن جدًا. | كود عام: SQL (Hive)، ML، بحث، image analysis… ونظام تشغيل للداتا بالمعنى الواسع (HBase، Impala). |
| الأعطال | لو نود وقعت، الكويري كلها تتعاد. (الكويريز قصيرة.) | يعيد الـ task بس، وبيكتب لديسك كتير. السبب الحقيقي: Google كانت بتشغل الـ batch كـ low priority ممكن يتقتل في أي وقت (preemption) لصالح الخدمات المهمة — فرص قتل task ساعة ≈ 5%. |
ما بعد MapReduce
مشكلة الـ Materialization
MapReduce بيكتب الـ output الوسيط (بين كل job والتاني) في HDFS ومنسوخ على كذا ماكينة. ده بطيء: job لازم يستنى كل الـ tasks اللي قبله تخلص، mappers مالهاش لازمة بتقرا وتعيد كتابة، وreplication لداتا مؤقتة. (Unix pipes بتعمل streaming!)
Dataflow Engines: Spark, Tez, Flink
بيتعاملوا مع الـ workflow كله كـ job واحد (DAG من الـ operators):
Fault tolerance: بما إن الوسيط مش متخزن، لو ماكينة وقعت بيعيد حسابه من آخر داتا متاحة (Spark: RDD lineage؛ Flink: checkpoints). المهم: الـ operators تبقى deterministic (احذر من: ترتيب hash table، random من غير seed، الوقت الحالي) — وإلا لو اتعاد الحساب هيطلع نتيجة مختلفة والـ downstream يتلخبط.
Graphs والمعالجة التكرارية
خوارزميات زي PageRank محتاجة تكرار لحد ما توصل (iterative). MapReduce وحش فيها (بيقرا الداتا كلها كل مرة). الحل: Bulk Synchronous Parallel (BSP) / Pregel model (Apache Giraph، Spark GraphX، Flink Gelly): كل vertex “بيفتكر” state وبيبعت رسايل لجيرانه على الـ edges في كل iteration. وfault tolerance بـ checkpoints دورية. المشكلة: الـ partitioning بيتعمل غالبًا عشوائي، والشبكة بتبقى bottleneck — لو الجراف يدخل في ماكينة واحدة، غالبًا هتبقى أسرع!
High-level APIs
Hive, Pig, Cascading, Crunch، وبعدين Spark و Flink عملوا APIs بـ relational operators (join، group by، filter). ميزتها مش بس كود أقل: الـ framework يقدر يختار join algorithm أوتوماتيك (cost-based optimizer) — بيقربوا من الـ declarative بتاع MPP مع الاحتفاظ بمرونة الكود العادي. وكمان column-oriented formats (Parquet) و vectorized execution. وفيه مكتبات متخصصة: ML (Mahout, MLlib)، k-nearest neighbors، genome analysis.
🧪 Lab
اعمل log analysis بأدوات Unix على أي access log عندك (nginx مثلًا):
cat access.log | awk '{print $7}' | sort | uniq -c | sort -rn | head -n 5بعدين اعمل نفس الحاجة بـ Spark (pyspark أو Spark SQL):
from pyspark.sql import SparkSessionspark = SparkSession.builder.getOrCreate()df = spark.read.text("access.log")df.selectExpr("split(value, ' ')[6] as url").groupBy("url").count().orderBy("count", ascending=False).show(5)