ما الذي يحدث فعليًا عند تشغيل مهمة في Spark؟

مقدمة

مع نمو مجموعات البيانات، قد تصبح معالجتها بكفاءة على جهاز واحد أمرًا صعبًا بسبب القيود المفروضة على سعة وحدة المعالجة المركزية والذاكرة ومعدل نقل البيانات في وحدات التخزين.

تعالج Apache Spark هذه التحديات من خلال توفير محرك موزع لمعالجة البيانات، يستطيع تقسيم أحمال العمل إلى أقسام وتنفيذ المهام بالتوازي عبر أنوية وحدة المعالجة المركزية وأجهزة متعددة.

لكن Spark لا تقتصر ببساطة على إضافة مزيد من الأجهزة.

يُعد فهم كيفية عمل مكوناتها معًا — برامج التشغيل والمنفذون وRDDs وDataFrames والأقسام والمهام والمراحل والمهام الفرعية — أمرًا ضروريًا لتصميم مسارات بيانات فعالة وموثوقة.

سأستعرض في هذا المقال بنية Spark وآليات تنفيذها وأهم اعتبارات الأداء، مستعينًا بمثال عملي لتجميع بيانات مبيعات التجارة الإلكترونية.

1. لماذا Apache Spark؟ — معالجة البيانات بالتوازي

Apache Spark محرك حوسبة موزعة مصمم لمعالجة البيانات وهندستها على نطاق واسع.

تعالج Spark البيانات الضخمة بالاستفادة من المعالجة المتوازية للبيانات:

الشكل أ. نظرة عامة على بنية عنقود Apache Spark الموزع (العنقود وبرنامج التشغيل وعقد العمال)، مع مقارنة التنفيذ على عقدة واحدة (أنشأته Kuriko IWAI)

بالمقارنة مع المعالجة على عقدة واحدة (يمين الشكل أ)، تقسم Spark العمل إلى مهام باستخدام عقد متعددة، وتنفذها عبر الموارد المتاحة، عند توفرها، على امتداد الأجهزة.

لا يعني التوزيع هنا أن استخدام أجهزة متعددة أمر غير إلزامي. إذ يمكن تشغيل Spark في الوضع المحلي على جهاز واحد، وحتى في هذه الحالة يمكن تقسيم البيانات ومعالجتها بالتوازي.

2. البنية الأساسية لـ Apache Spark

تتكون Spark من عدة مكونات أساسية:

• برنامج تشغيل Spark (برنامج التشغيل): عملية التنسيق المركزية لتطبيق Spark.
• مدير العنقود: خدمة خارجية مسؤولة عن تخصيص موارد الأجهزة، مثل أنوية وحدة المعالجة المركزية والذاكرة.
• عقد العمال (عامل): جهاز مادي أو افتراضي، أو عقدة حوسبة في العنقود.
• المنفذون: عمليات عمال موزعة تُشغَّل على عقد العمال.
• Java Virtual Machine (JVM): بيئة التشغيل اللازمة لتنفيذ الشيفرة الوسيطة لـ Spark (بما أن Spark مكتوبة بلغتي Scala وJava).

أولًا، ينسق برنامج التشغيل تطبيق Spark ويحوّله إلى رسم بياني موجّه لا دوري (DAG) لمهام التنفيذ الفعلية، ثم يوزع هذه المهام على المنفذين.

بعد ذلك، يخصص مدير العنقود الموارد وفق نظام إدارة الموارد في بيئة النشر.

توفر عقد العمال موارد الحوسبة للمنفذين، الذين ينفذون المهام ويحتفظون بالبيانات أثناء تنفيذها.

تستخدم بعض البيئات المُدارة، مثل Databricks، منفذًا واحدًا لكل عامل، لكن هذه ليست قاعدة عامة في Spark. إذ يمكن أن يستضيف العامل عدة منفذين، بحسب إعدادات العنقود.

2-1. حالة استخدام في التجارة الإلكترونية — بنية بيانات الإدخال

لنتخيل شركة صغيرة متخصصة في التجارة الإلكترونية.

لديها 10,000 سجل مبيعات، يتضمن كل منها وحدة حفظ مخزون (SKU) وكمية مباعة. لنفترض وجود ثلاث وحدات SKU محتملة: A, B، و C.

الجدول 1. مثال على توزيع سجلات مبيعات التجارة الإلكترونية والتجميعات المتوقعة.

هدفنا هو حساب إجمالي الوحدات الصالحة المباعة لكل وحدة SKU:

1. استبعاد السجلات التي qty <= 0، ثم
2. تجميع العينات حسب وحدة SKU.

في مجموعة البيانات الكاملة التوضيحية، لنفترض أن 9,800 من أصل 10,000 سجل تبقى بعد التصفية. وإليك الإجماليات النهائية المتوقعة من Spark:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. عنقود Spark

سأحدد عنقودًا مبسطًا لحالة الاستخدام لدينا.

عقد العمال (العمال)

سأحدد 3 عقد عمال، لكل منها نوع مثيل يضم 8 وحدات معالجة مركزية افتراضية (vCPU) وذاكرة وصول عشوائي (RAM) بسعة 32 GiB:

الشكل ب. توزيع الموارد على عقد العمال (أنشأته Kuriko IWAI)

يتوفر في النظام إجمالًا 24 وحدة معالجة مركزية افتراضية (vCPU).

منفذو Spark وتخصيص المعالج والذاكرة

لنفترض أنني سأخصص:

• 7 منفذين.
• نواتي vCPU لكل منفذ.
• ذاكرة كومة JVM بسعة 3GiB لكل منفذ.

يطلب تطبيق Spark من مدير العنقود تخصيص الموارد لهؤلاء المنفذين السبعة.

بعد ذلك، يحدد مدير العنقود تخصيصات المعالج والذاكرة استنادًا إلى استخدام الموارد لدى كل عقدة عاملة.

على سبيل المثال، يقرر إنشاء 14 خانة مهام — نواتا vCPU لكل منفذ × 7 منفذين:

الشكل C-1. أنوية vCPU والذاكرة وتوزيع المنفذين على العقد العاملة، مع اختلال توزيع الموارد في العقدة العاملة A

لاحظ أن سعة الذاكرة البالغة 3 GiB المخصصة للمنفذ ليست إجمالي ذاكرته، إذ عليه تخصيص ذاكرة إضافية للنفقات التشغيلية للعقد العاملة ولعمليات التنفيذ الأخرى.

على افتراض أن المهمة الواحدة تتطلب 1 vCPU (spark.task.cpus = 1)، تخصص كل عقدة عاملة وحدات vCPU لديها:

الشكل C-2. التوزيع المتوازي لوحدات vCPU على العقد العاملة

يوضح هذا نقطتين أساسيتين:

1. لكل عقدة عاملة معدل استخدام خاص للمعالج (في الشكل C-2، يبلغ الاستخدام في العقدة العاملة A نسبة 75%، بينما يظل عند 50% في العقدتين العاملتين B وC).
2. على الرغم من أن النظام يضم 24 نواة vCPU (8 وحدات vCPU لكل عقدة عاملة × 3 عقد عاملة)، فإنه لا يستطيع استخدام سوى 14 نواة vCPU؛ أما الوحدات العشر المتبقية فتبقى غير مستخدمة لهذه المهام.

2-3. الآلة الافتراضية لجافا (JVM) وراء Spark

بُني Apache Spark على الآلة الافتراضية لجافا (JVM).

JVM بيئة تشغيل لازمة لتنفيذ الشيفرة الوسيطة لـ Spark.

بما أن Spark مكتوب بلغة Scala/Java:

• عند استخدام Scala أو Java في التطبيق: تعمل الشيفرة محليًا داخل JVM.
• عند استخدام PySpark (Python): تنفذ عملية Python الشيفرة، لكنها تتواصل مع برنامج تشغيل يستند إلى JVM عبر جسر مثل Py4J. ولا تزال منفذات Spark على العقد العاملة تعمل داخل JVM، الذي يدير بدوره عمليات عاملة لـ Python لتنفيذ شيفرة Python والدوال المعرّفة من المستخدم في Python (Python UDFs).

في كلتا الحالتين، تدير JVM خيوط التنفيذ وتخصيص الذاكرة وجمع القمامة لكل من برنامج التشغيل والمنفذات.

وحدات JVM لبرنامج التشغيل والمنفذات في تطبيق PySpark

في تطبيق PySpark، يتواصل برنامج التشغيل في Python مع برنامج تشغيل Spark الذي يعمل داخل JVM؛

• برنامج التشغيل في Python: تُكتب فيه شيفرة PySpark على مستوى التطبيق وتُنفذ.
• JVM برنامج التشغيل: ينسق تنفيذ Spark وجدولته.
• JVM المنفذ: ينفذ مهام Spark ويدير موارد التنفيذ. وتُستخدم عمليات عاملة لـ Python عندما تتطلب المهام تنفيذ شيفرة Python.
• JVM العميل (عملية العميل أو جهاز إرسال المهام): JVM يعمل على الجهاز المحلي أو على عقدة طرفية/بوابة.

كما يوضح الشكل، يتصل JVM العميل أولًا بـJVM برنامج التشغيل، الذي ينسق بعد ذلك مع وحدات JVM للمنفذات:

الشكل D. العلاقة العامة بين وحدات JVM الخاصة بالعميل وبرنامج التشغيل والمنفذات

من هنا، يستضيف JVM برنامج التشغيل Spark Driver لإدارة التطبيق، بينما تستضيف وحدات JVM الخاصة بالمنفذات المنفذات الفردية المسؤولة عن تنفيذ تلك المهام.

تضم ذاكرة كومة JVM الخاصة بكل برنامج تشغيل مساحة لمعالجة البيانات والتخزين المؤقت وكائنات المستخدم بكفاءة.

2-4. نظرة معمقة على مجموعات البيانات الموزعة القابلة للاستعادة (RDDs)

مجموعة البيانات الموزعة القابلة للاستعادة (RDD) هي مجموعة غير قابلة للتغيير ومقسمة إلى أجزاء من العناصر، يمكن لـ Spark معالجتها بالتوازي.

تتتبع RDDs التبعيات بين التحويلات، ما يتيح لـ Spark إعادة احتساب الأجزاء المفقودة في كثير من حالات الإخفاق.

يعكس الاسم ثلاث خصائص:

• قابلة للاستعادة: يسجل Spark سلسلة النسب —أي RDD يعتمد على أي RDD سابق— وغالبًا ما يستطيع إعادة بناء الأجزاء المفقودة بدلًا من الاحتفاظ بنسخ متماثلة من كل نتيجة وسيطة.
• موزعة: تُمثَّل مجموعة البيانات في صورة أجزاء قد يعالجها Spark على منفذات مختلفة. ولا يشترط استخدام عدة أجهزة فعلية؛ إذ يدعم الوضع المحلي RDDs أيضًا.
• مجموعة بيانات: يمكن أن تحتوي RDD على أزواج مفتاح-قيمة وأعداد صحيحة وسلاسل نصية وصفوف وعناصر أخرى مدعومة، وليس على صفوف علائقية فحسب. وتتضمن واجهات RDD البرمجية عمليات مثل map, filter, reduceByKey, join، و aggregateByKey.

إن RDD هو تجريد منطقي لمجموعة بيانات وعمليات حوسبتها، وليس إعلانًا بأن أقسامًا بعينها مخصصة دائمًا لعمّال محددين.

الأقسام

القسم هو جزء منطقي من البيانات، ويقابل عادةً مهمة واحدة ضمن مرحلة.

لنفترض أننا أنشأنا 100 قسم (اختيرت هذه الأقسام الأولية المئة صراحةً لأغراض الشرح).

مع وجود 10,000 عنصر مصدر، ينتج عن ذلك نحو 100 عنصر في كل قسم إذا وُزّعت المدخلات توزيعًا متقاربًا:

الشكل E. التوزيع المتوازي للأقسام على 100 تقسيم أولي للبيانات

3. واجهة RDD مقابل واجهة DataFrame: التحسين الداخلي باستخدام Catalyst

يساعد مثال RDD الأصلي على فهم النسب، والأقسام، وعمليات إعادة التوزيع.

لكن في عمليات استخراج البيانات وتحويلها وتحميلها (ETL) المنظّمة المعتادة، تُعد واجهة DataFrame نقطة البداية المفضلة لأنها توفر أعمدة ومعلومات عن المخطط وتعبيرات شبيهة بـSQL وعمليات يطّلع عليها المحسِّن.

يوضح المخطط أدناه العلاقة:

الشكل F. العلاقة بين مفاهيم الرسوم البيانية في Spark؛ ونسب RDD وDataFrame ورسم التنفيذ البياني غير الدوري الموجّه (DAG) في Spark

مثلما يحدد نسب RDD عمليات التحويل، تمثل خطة استعلام DataFrame الاستعلام المنظّم الخاص به.

ينشئ Spark مهام التنفيذ استنادًا إلى التبعيات والمشغّلات المادية.

قد يستخدم تنفيذ DataFrame آليات قائمة على RDD داخليًا، لكن هذا لا يعني أن على المستخدمين إنشاء نسب RDD يدويًا عند كتابة شيفرة DataFrame.

3-1. خطة استعلام DataFrame

خطة استعلام DataFrame هي تسلسل الخطوات المنطقية والمادية التي ينشئها محرك التنفيذ (المربع الأرجواني في الشكل L).

تمر استعلامات DataFrame بعدة تمثيلات:

1. الخطة المنطقية: وصف للعمليات المطلوبة (الفحص والتصفية والتجميع والتجميع الحسابي).
2. الخطة المنطقية المحلّلة والمحسّنة: يحل Spark المراجع ويطبّق التحسينات المنطقية المناسبة، مثل تقليص الأعمدة أو دفع الشروط إلى مصدر البيانات.
3. الخطة المادية: يختار Spark مشغّلات قابلة للتنفيذ، مثل الفحص والتجميع الجزئي أو النهائي والتبادل والضمّ والفرز.
4. التنفيذ: تُشغَّل الخطة المادية باستخدام آليات التنفيذ في Spark، وتتضمن في نهاية المطاف مهامًا على المنفّذات.

في حالة الاستخدام لدينا، قد تتضمن الخطة المادية الخطوات التالية:

• التجميع الجزئي باستخدام التجزئة،
• التبادل بحسب SKU، و
• التجميع النهائي باستخدام التجزئة.

3-2. مقارنة العمليات: واجهة RDD مقابل واجهة DataFrame

لنقارن بين واجهتي RDD وDataFrame، مستخدمين تجميع SKU مثالًا.

واجهة RDD

العناصر هي (sku, qty) أزواج. ومنطق التجميع في Python صريح، ويتبع Spark تبعيات RDD:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

واجهة DataFrame

افترض أن sales_df لديه أعمدة sku (نصي) و qty (رقمي):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

في هذه الحالة، يحل المثالان المشكلة التجارية نفسها:

تصفية سجلات المبيعات غير الصالحة وتجميع الكميات الصالحة بحسب SKU.

وفي كلتا الحالتين، collect() يؤدي ذلك إلى تشغيل Spark وإعادة جميع السجلات الناتجة إلى برنامج التشغيل.

يكمن الفرق الأساسي هنا في تمثيل النتائج:

• واجهة RDD: تعيد عناصر Python، مثل (sku, total_qty) كائنات tuple.
• واجهة DataFrame: تعيد كائنات منظّمة Row ذات حقول مسماة، مثل sku و total_qty.

فيما يلي مقارنة شاملة بين واجهتي RDD وDataFrame:

الجدول 2. مقارنة الميزات ومستوى التجريد بين RDD API وDataFrame API: (إعداد Kuriko IWAI)

4. خلاصة

لا يتمثل الهدف من ضبط Spark في زيادة عدد الأقسام أو استخدام وحدة المعالجة المركزية كلٌّ على حدة إلى أقصى حد.

بل يتمثل في خفض تكلفة التنفيذ وزمن استجابته من البداية إلى النهاية، من دون التسبب في عدم الاستقرار.

في هذا المقال، نتعرّف إلى العلاقات الأساسية في نظامه:

• الموارد: تستضيف العُقد العاملة المنفّذات؛ وتشغّل المنفّذات المهام؛ ويحدّد إعداد وحدة المعالجة المركزية والذاكرة مقدار التزامن.
• البرمجة باستخدام RDD: تنتج التحويلات تعريفات غير قابلة للتغيير لـ RDD وسلسلة نسبها؛ وتطلق الإجراءات عملية الحساب.
• البرمجة باستخدام DataFrame: تنتج التحويلات عمليات استعلام منطقية؛ ويختار التحسين مشغّلات فعلية قابلة للتنفيذ؛ وتطلق الإجراءات عملية الحساب.
• التنفيذ: ينظّم Spark العمل في مهام ومراحل ومهمات، وتعالج المهمات الأقسام على المنفّذات. وتُنشئ عمليات تبديل البيانات وغيرها من التبعيات حدودًا تنسيقية بين المراحل.

لا يعد فهم كيفية تنفيذ Spark لأعباء العمل الموزعة سوى أساسٍ أولي.

يمكنك التعرّف إلى تدفّق حسابي من سبع خطوات باستخدام RDD API.