الوحدة 4 — قراءة البيانات والصيغ العمودية

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

CSV: النموذج البسيط والمكلف

CSV نصّ خام. لا مخطَّط، لا ضغط داخليّ، لا فهرسة. يفتح Spark الملفّ ويقرؤه سطرًا سطرًا، محلّلًا كلّ حقل يوقف مصادفةَ فاصلة أو سطر جديد. هذا ثمنه لا يُستهان به: على تاريخ الرحلات (60 مليون سطر، 30 عمودًا) بحجم 25 غيغابايت مضغوطة بـ gzip، تستغرق قراءة spark.read.csv عدّة دقائق فقط في التحميل والتحليل.

الأدهى أنّ Spark يستنتج المخطَّط افتراضيًّا. أضف inferSchema=True وسيمرّ Spark مرّتين على الملفّ: مرّة لتخمين نوع كلّ عمود، وأخرى لقراءته فعلًا. على 25 غيغابايت هذا وحده يزيد الزمن بنحو الضعف. الحلّ: صرِّح بالمخطَّط:

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType

schema = StructType([
    StructField("annee", IntegerType()),
    StructField("mois", IntegerType()),
    StructField("compagnie", StringType()),
    StructField("origine", StringType()),
    StructField("destination", StringType()),
    StructField("distance", DoubleType()),
    StructField("retard_arrivee", DoubleType()),
])

vols = spark.read.schema(schema).csv("s3a://donnees-vols-brut/*.csv.gz",
                                     header=True)

بمخطَّط صريح تُقرأ الملفّات مرّة واحدة، ويرفض Spark فورًا الأسطر التي لا تتَّفق مع الأنواع بدل بلع الأخطاء.

Parquet: الصيغة العموديّة

Parquet صيغة ملفّ عموديّة ومُضغَّطة صمَّمها فريق تويتر وكلاوديرا لِلبيانات التحليليّة. لا تُخزَّن الأسطر متتاليةً كما في CSV، بل يُجمَع كلّ عمود على حدة، ويضغَط بخوارزميّة تناسبه (Snappy، Zstandard، LZ4). لِكلّ عمود إحصاءات مصغَّرة (حدّ أدنى، أعلى، عدد القيم الفارغة) تعبِّئ ملفّ الأدلّة الوصفيّة.

ثلاث نتائج مباشرة:

• الضغط أقوى: العمود الواحد قيم متشابهة، فتضغطها الخوارزميّة إلى ربع حجم CSV تقريبًا.
• تقليم الأعمدة: إذا طلبت ثلاثة أعمدة، لا يقرأ Spark إلّاها. Parquet ينظّم البيانات بحيث يمكن قفز البقيّة.
• دفع المُرشِّحات: يستخدم قارئ Parquet الإحصاءات لتخطّي كتل كاملة لا تحوي أيّ سطر يستوفي الشرط. filter("annee = 2019") يقرأ فقط الكتل التي تعبر النطاق.

على مشروعنا:

vols = spark.read.parquet("s3a://donnees-vols/")

لا حاجة لتحديد المخطَّط: Parquet يحمله في رأسه. القراءة عادةً أسرع من CSV بعشرة أضعاف أو أكثر.

التقسيم بعمود

يمكن كتابة Parquet مقسَّمًا بعمود (أو أكثر): تُخزَّن كلّ قيمة مميّزة في مجلَّد فرعيّ:

s3a://donnees-vols/
  annee=2015/mois=01/part-00000.parquet
  annee=2015/mois=02/part-00000.parquet
  ...
  annee=2023/mois=12/part-00013.parquet

عندما تكتب filter("annee = 2019")، لا يفتح Spark إلّا مجلَّد annee=2019. هذا تقليم الأقسام (partition pruning)، وهو أقوى من دفع المُرشِّحات لأنّه يعمل على مستوى الملفّات كلّها لا الكتل.

نكتب البيانات مقسَّمة مرّة، ونقرؤها بمرشِّح مقسَّم عشرات المرّات:

# كتابة أوّليّة، مرّة واحدة
vols.write.partitionBy("annee", "mois").parquet("s3a://donnees-vols/")

# قراءة يوميّة، سريعة
donnees_2019 = spark.read.parquet("s3a://donnees-vols/").filter("annee = 2019")

اختيار عمود التقسيم يخضع لقاعدة بسيطة: قِيَم قليلة نسبيًّا وتوزيع متوازن. annee مثاليّ (10 قيم، عدد رحلات متقارب). id_rihla كارثيّ (60 مليون قيمة، كلّ مجلَّد يحوي سطرًا واحدًا). التقسيم المبالغ فيه ينتج مشكلة الملفّات الصغيرة التي تُبطّئ كلّ قراءة لاحقة.

أنواع بيانات أخرى

• JSON: مرن لكنّه ثقيل، بلا مخطَّط، مضغوط بسوء. مقبول لأنابيب الاستهلاك، لا للتخزين التحليليّ.
• ORC: عموديّ أيضًا، بديل شائع لـParquet في عالَم Hive. Spark يقرأه ويكتبه، لكنّ الاختيار في المجتمع مال بوضوح إلى Parquet.
• Delta Lake: طبقة فوق Parquet تضيف المعاملات ونسخ الاستعادة والتحديث الموضعيّ. عليها تُبنى معظم مخازن البيانات الحديثة.

قاعدة عمليّة للتخزين

اقرأ من CSV إذا كان مصدر البيانات لا يعرف غيره. اكتب من جانبك دائمًا إلى Parquet مقسَّمًا بعمود ذي دلالة زمنيّة أو تصنيفيّة قليلة القيم. بذلك تدفع تكلفة التحويل مرّة واحدة، ثمّ يستفيد كلّ استعلام لاحق من الضغط وتقليم الأقسام.

الخلاصة

• CSV بسيط لكنّه بلا مخطَّط ولا فهرسة؛ استنتاج المخطَّط يعني قراءتين، فصرِّح دائمًا بالمخطَّط.
• Parquet صيغة عموديّة مضغوطة تدعم دفع المُرشِّحات وتقليم الأعمدة؛ أسرع من CSV بعشرة أضعاف عادةً.
• التقسيم بعمود ينشئ مجلَّدات فرعيّة تسمح بتقليم الأقسام؛ اختر عمودًا قليل القيم ومتوازنًا.
• Delta Lake يضيف المعاملات إلى Parquet وصار البنية التحتيّة الافتراضيّة لخزائن البيانات الحديثة.

الوحدة التالية: من DataFrame إلى Spark ML — كيف تُعِدّ متغيّراتك بمحوّلات (VectorAssembler، StringIndexer، OneHotEncoder) قبل تمريرها إلى مقدِّر.