بحيرات البيانات وهندسة Lakehouse

نموذج lakehouse يجمع بين مرونة بحيرات البيانات وموثوقية مستودعات البيانات. فهم هذا التطور حاسم لمقابلات هندسة البيانات الحديثة.

تطور بحيرة البيانات

تحديات بحيرة البيانات التقليدية

التحدي	التأثير
لا معاملات ACID	فساد البيانات، قراءات غير متسقة
فرض المخطط	قمامة داخل، قمامة خارج
لا فهرسة	مسح كامل لكل استعلام
بيانات وصفية قديمة	تخطيط استعلام بطيء
لا سفر عبر الزمن	لا يمكن التعافي من الكتابات السيئة

حل Lakehouse

┌─────────────────────────────────────────────────────┐
│                  محركات الاستعلام                   │
│     (Spark, Trino, Dremio, Athena, Databricks SQL)  │
├─────────────────────────────────────────────────────┤
│              طبقة صيغة الجدول                       │
│         (Delta Lake, Apache Iceberg, Hudi)          │
├─────────────────────────────────────────────────────┤
│              تخزين الكائنات                          │
│           (S3, GCS, ADLS, MinIO)                    │
└─────────────────────────────────────────────────────┘

Delta Lake

صيغة lakehouse مفتوحة المصدر من Databricks.

الميزات الرئيسية

الميزة	الفائدة
معاملات ACID	كتابات متزامنة موثوقة
تطور المخطط	إضافة/إعادة تسمية الأعمدة بدون إعادة كتابة
السفر عبر الزمن	استعلام أي إصدار سابق
Z-ordering	تجميع متعدد الأبعاد
تغذية تغيير البيانات	بث التغييرات للمصب

العمليات الشائعة

# إنشاء جدول Delta
df.write.format("delta").save("/data/events")

# السفر عبر الزمن - قراءة إصدار سابق
spark.read.format("delta") \
    .option("versionAsOf", 5) \
    .load("/data/events")

# السفر عبر الزمن - قراءة بالطابع الزمني
spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-01") \
    .load("/data/events")

# تحسين و Z-order
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, "/data/events")
delta_table.optimize().executeZOrderBy("user_id", "event_date")

# تنظيف الإصدارات القديمة (احتفظ 7 أيام)
delta_table.vacuum(168)  # ساعات

سؤال المقابلة: "كيف يحقق Delta Lake معاملات ACID؟"

الجواب: "Delta Lake يستخدم سجل معاملات (_delta_log/) يسجل كل تغيير كـ JSON commits. الكتابات ذرية من خلال التزامن المتفائل—كل كتابة تنشئ ملف commit جديد. القراءات تستخدم السجل لتحديد أي ملفات تقرأ. التعارضات تُكتشف بالتحقق إذا كانت المعاملات الملتزمة تتداخل مع الكتابة الحالية."

Apache Iceberg

صيغة طورتها Netflix، الآن مشروع Apache.

الميزات الرئيسية

الميزة	الفائدة
التقسيم المخفي	تطور التقسيم بدون إعادة كتابة
تطور المخطط	دعم كامل (إضافة، حذف، إعادة تسمية، إعادة ترتيب)
السفر عبر الزمن	إصدار قائم على اللقطات
تطور التقسيم	تغيير التقسيم بدون هجرة
محركات متعددة	Spark, Flink, Trino, Hive, Presto

الهندسة

┌─────────────────────────────────────┐
│          الكتالوج (البيانات الوصفية)│
│  (Hive Metastore, Glue, Nessie)     │
├─────────────────────────────────────┤
│        طبقة البيانات الوصفية        │
│   ┌──────────────────────────────┐  │
│   │ قائمة البيان (لقطة)          │  │
│   │   └── ملفات البيان           │  │
│   │       └── مراجع ملفات البيانات│  │
│   └──────────────────────────────┘  │
├─────────────────────────────────────┤
│          ملفات البيانات             │
│    (Parquet, ORC, Avro)             │
└─────────────────────────────────────┘

العمليات الشائعة

-- إنشاء جدول Iceberg
CREATE TABLE events (
    event_id BIGINT,
    event_time TIMESTAMP,
    user_id INT,
    event_type STRING
)
USING iceberg
PARTITIONED BY (days(event_time), bucket(16, user_id));

-- السفر عبر الزمن
SELECT * FROM events VERSION AS OF 123456;
SELECT * FROM events TIMESTAMP AS OF '2024-01-01 00:00:00';

-- تطور التقسيم (لا إعادة كتابة بيانات!)
ALTER TABLE events ADD PARTITION FIELD bucket(8, region);

-- انتهاء صلاحية اللقطات
CALL system.expire_snapshots('events', TIMESTAMP '2024-01-01 00:00:00');

سؤال المقابلة: "ما ميزة التقسيم المخفي في Iceberg؟"

الجواب: "مع التقسيم المخفي، المستخدمون يكتبون استعلامات باستخدام قيم الأعمدة الفعلية (WHERE event_time > '2024-01-01')، و Iceberg يترجم هذا تلقائياً لمسند التقسيم الصحيح. هذا يعني:

1. تطور التقسيم لا يتطلب تغييرات الاستعلام
2. لا حاجة لمعرفة مخطط التقسيم
3. تحويلات التقسيم (يوم، شهر، bucket) مجردة"

Apache Hudi

صيغة طورتها Uber للمعالجة التدريجية.

الميزات الرئيسية

الميزة	الفائدة
Upserts	دعم أصلي لتحديثات مستوى السجل
استعلامات تدريجية	قراءة السجلات المتغيرة فقط
فهرسة مستوى السجل	بحث وتحديثات سريعة
الضغط	دمج الملفات الصغيرة تلقائياً
أنواع الجداول	Copy-on-Write مقابل Merge-on-Read

مقارنة أنواع الجداول

الجانب	Copy-on-Write (CoW)	Merge-on-Read (MoR)
تأخير الكتابة	أعلى (إعادة كتابة الملفات)	أقل (إلحاق السجلات)
تأخير القراءة	أقل (مدمج مسبقاً)	أعلى (دمج عند القراءة)
الأفضل لـ	أعباء العمل الثقيلة القراءة	أعباء العمل الثقيلة الكتابة
التخزين	أكثر (إعادة كتابة كاملة)	أقل (التغييرات فقط)

مقارنة الصيغ

الميزة	Delta Lake	Iceberg	Hudi
الأصل	Databricks	Netflix	Uber
الاستخدام الأساسي	lakehouse عام	التحليلات	CDC/البث
ACID	✅	✅	✅
السفر عبر الزمن	✅	✅	✅
تطور المخطط	جيد	ممتاز	جيد
تطور التقسيم	محدود	ممتاز	محدود
Upserts	جيد	جيد	ممتاز
القراءات التدريجية	Change Data Feed	تدريجي	أصلي
دعم المحرك	أفضل Spark	متعدد المحركات	جيد Spark

سؤال المقابلة: "كيف تختار بين Delta و Iceberg و Hudi؟"

إطار الإجابة:

إذا تحتاج	اختر
نظام Databricks البيئي	Delta Lake
قابلية نقل المحرك	Iceberg
CDC/upserts ثقيل	Hudi
عمليات بسيطة	Delta Lake
مرونة التقسيم	Iceberg
استيعاب في الوقت الحقيقي	Hudi

هندسة Medallion

نمط تصميم lakehouse شائع.

┌───────────┐    ┌───────────┐    ┌───────────┐
│  Bronze   │───▶│  Silver   │───▶│   Gold    │
│  (خام)    │    │ (منظف)   │    │  (مجمع)   │
└───────────┘    └───────────┘    └───────────┘

الطبقة	الغرض	المخطط	الجودة
Bronze	استيعاب خام	مخطط عند القراءة	كما هو من المصدر
Silver	منظف، موحد	مفروض	موثق، مزال التكرار
Gold	تجميعات الأعمال	منمذج	جاهز لـ BI

مثال التنفيذ

# Bronze: استيعاب خام
raw_df = spark.read.json("/landing/events/")
raw_df.write.format("delta").mode("append") \
    .save("/bronze/events")

# Silver: تنظيف وإزالة التكرار
bronze_df = spark.read.format("delta").load("/bronze/events")
silver_df = bronze_df \
    .dropDuplicates(["event_id"]) \
    .filter(col("user_id").isNotNull()) \
    .withColumn("event_date", to_date("event_time"))
silver_df.write.format("delta").mode("overwrite") \
    .partitionBy("event_date") \
    .save("/silver/events")

# Gold: تجميعات الأعمال
silver_df = spark.read.format("delta").load("/silver/events")
gold_df = silver_df.groupBy("event_date", "event_type") \
    .agg(count("*").alias("event_count"))
gold_df.write.format("delta").mode("overwrite") \
    .save("/gold/daily_event_summary")

نظرة المقابلة: lakehouse يمثل تقارب بحيرات البيانات والمستودعات. فهم متى تستخدم lakehouse مقابل المستودع التقليدي يُظهر نضج معماري.

بعد ذلك، سنغوص في الأبعاد المتغيرة ببطء وتنفيذاتها. :::