انتقل إلى المحتوى الرئيسي

الوحدة 2 — RDD و DataFrame و Dataset

تعرض Spark ثلاث تجريدات لتمثيل البيانات، لِكلٍّ منها تاريخ وقيود ومكاسب. تُقرأ هذه الوحدة بعد الوحدة 1 كتمييزٍ مباشر: الأقسام والمنفّذون هم أنفسهم، لكنّ ما يجري بينهم يتغيّر تغيُّرًا جذريًّا حسب التجريد المُختار.

RDD: الجيل الأوّل​

RDD أو Resilient Distributed Dataset هي التجريد التأسيسي منذ 2010. تجمع مجموعة من الكائنات موزَّعة على الأقسام، مع ضمان المرونة: إن سقط منفّذ يُعاد حساب أقسامه من مصدرها لأنّ Spark يحفظ سلسلة العمليّات التي أنشأتها. يعرض RDD واجهةً برمجيّة صريحة: map، filter، reduce، groupByKey. يجرى الحساب حرفيًّا كما تكتبه.

هذه الصراحة تحدُّها: لأنّ Spark يُعامِل كائناتك ككتلة معتمة (نقلها بالتسلسل عبر Pickle في بايثون، جافا Serialization في JVM)، لا يعرف شيئًا عن أنواع الحقول ولا عن الفرز الأنسب. كلّ تحسين تفعله يدويًّا. ويكفي المثال التالي لِيرى المرء الفرق:

rdd = spark.sparkContext.textFile("s3a://donnees-vols/2018.csv")
retards = rdd.map(lambda l: l.split(","))\
             .filter(lambda c: float(c[10]) > 15)\
             .count()

لا يعرف Spark أنّ العمود العاشر رقميّ، ولا أنّ 90 % من الأسطر ستُقصى. سيقرأ الملفّ كلّه ويُحلّله سطرًا سطرًا في بايثون، ويعبر البيانات بالكامل بين JVM و بايثون.

DataFrame: تجريد بمخطَّط​

في 2015 قدَّمت Spark DataFrame: جدول من الأسطر (Row) بأعمدة مسمّاة ونوع لكلّ عمود. الفرق عن RDD بنيويّ: تعرف Spark الآن مخطَّط البيانات، فتُمرّر الاستعلامات إلى محسِّن Catalyst الذي يُعيد كتابتها قبل التنفيذ. المثال نفسه يصير:

vols = spark.read.parquet("s3a://donnees-vols/2018/")
retards = vols.filter("retard_arrivee > 15").count()

بضعة أسطر، لكنّ Catalyst يقوم بأشياء لم تطلبها:

  • دفع المُرشِّح (predicate pushdown): يوصل شرط retard_arrivee > 15 إلى قارئ Parquet نفسه، الذي يستخدم الإحصاءات لِتخطّي كتل كاملة لا تحوي أيّ سطر يستوفي الشرط.
  • تقليم الأعمدة: لا يقرأ إلّا العمود المطلوب من بين ثلاثين عمودًا.
  • إعادة ترتيب الانضمامات والاختيار بين خوارزميّاتها (broadcast join مقابل sort-merge join).
  • توليد شيفرة Java فيما يُعرف بـ Whole-Stage Codegen، فيصير التنفيذ قريبًا من C.

النتيجة: على البيانات نفسها يمكن أن يكون DataFrame أسرع من RDD بعشرة أضعاف أو أكثر، ولم تلمس خوارزميّتك.

Dataset: النموذج المطبوع​

في Scala وJava تعرض Spark تجريدًا رابعًا هو Dataset[T]: DataFrame مع ضمان نوعي فحوصه المُترجِم. تُكتب dataset.filter(v => v.retardArrivee > 15) وتخطئ عند البناء لا عند التشغيل إن أخطأت اسم الحقل. في PySpark لا يوجد Dataset: بايثون ديناميكي فلا يفيد التطبيع بأنواع مُترجَمة.

Spark SQL​

فوق DataFrame تعرض Spark تنفيذًا كاملًا لـSQL. تُسجّل DataFrame كجدول مؤقّت وتستعلم عنه:

vols.createOrReplaceTempView("vols")
resultat = spark.sql("""
  SELECT compagnie, AVG(retard_arrivee) AS moy
  FROM vols
  WHERE annee = 2018
  GROUP BY compagnie
  ORDER BY moy DESC
""")

الاستعلام يمرّ بنفس Catalyst ويعطي نفس الأداء. الميزة الأهمّ ليست تقنيّة: SQL معروف في فِرَق البيانات، وموحَّد بين DataFrame و SQL يعني أنّ محلّل بيانات يكتب استعلامًا يعمل كما يعمل الأنبوب البرمجيّ لِعالِم البيانات، حرفيًّا.

متى تعود إلى RDD

لا تعود إلى RDD طلبًا للسرعة: DataFrame أسرع. تعود إلى RDD حين تحتاج تحكّمًا لا يُعبِّر عنه SQL بسهولة: خوارزميّة رسم بيانيّ مخصَّصة، تحكّم حبيب في التقسيم لغرض بحثي، تسلسل ثنائي مع مكتبة خارجيّة، أو حين تُعالج بيانات لا مخطَّط لها ولا يمكن استنتاجه. في تسعة وتسعين بالمئة من مشاريع MLlib، لن تكتب RDD بيدك.

الخلاصة​

  • RDD تجريد صريح بلا مخطَّط: مرن لكنّه أعمى عن التحسينات، وأبطأ في الاستعمال العادي.
  • DataFrame جدول بمخطَّط يمرّ بمحسِّن Catalyst: هو التجريد الافتراضي لكلّ عمل مع MLlib.
  • Dataset يضيف الأمان النوعي في Scala وJava؛ لا وجود له في PySpark.
  • Spark SQL يعرض SQL كاملًا فوق DataFrame بنفس الأداء، ويُوحِّد لغة الفريق حول جدول مشترك.

الوحدة التالية: التقييم الكسول، التحويلات والإجراءات، وكيف يقرأ المرء خطّة تنفيذ Spark قبل تشغيلها.