قراءة وتحليل ملفات ضخمة (بحجم جيجابايت) في ثوانٍ باستخدام PySpark DataFrames

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

إذا كنت قد قرأت مقال ما هو Apache Spark؟ ولماذا تتوقف مكتبة Pandas عن العمل مع البيانات الضخمة (Big Data)؟ فستعرف أن الفكرة الأساسية ليست مجرد “فتح ملف كبير”، بل تنفيذ التحليل على أجزاء متعددة من البيانات بالتوازي. كما أن إعداد البيئة بالشكل الصحيح يبدأ من إعداد بيئة PySpark: معالجة البيانات الموزعة على عدة أجهزة في نفس الوقت قبل الانتقال إلى العمل الإنتاجي.

لماذا DataFrame في PySpark أسرع من الأدوات التقليدية؟

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

ميزة أخرى مهمة هي التحسين المنطقي والفيزيائي للاستعلامات عبر Catalyst Optimizer ومحرك التنفيذ Tungsten. لذلك فإن كتابة أوامر بسيطة مثل التصفية والتجميع قد تتحول داخلياً إلى خطة تنفيذ عالية الكفاءة دون تدخل يدوي كبير.

في معمارية البيانات الحديثة، السرعة لا تأتي فقط من قوة الخادم، بل من اختيار تنسيق تخزين مناسب، وتقليل حجم البيانات المقروءة، وتوزيع العمل على Partitions متوازنة، ثم تنفيذ التحويلات بأسلوب كسول Lazy Evaluation.

بناء جلسة SparkSession وقراءة ملف ضخم

أول خطوة عملية هي إنشاء جلسة تشغيل مع تحديد بعض الإعدادات الأساسية مثل عدد الأنوية وحجم الذاكرة وسياسة ضغط السجلات. سواء كان الملف بصيغة CSV أو JSON أو Parquet، فإن طريقة القراءة يجب أن تكون واعية بنوعية المخطط وخصائص الملف.

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("LargeFileAnalysis")
    .master("local[*]")
    .config("spark.sql.shuffle.partitions", "16")
    .config("spark.driver.memory", "4g")
    .getOrCreate()
)

df = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("data/large_sales_file.csv")
)

print(df.count())
df.printSchema()
df.show(5, truncate=False)

السطر الخاص بـ inferSchema مريح للتجارب السريعة، لكنه ليس الأفضل في البيئات الإنتاجية، لأن استنتاج الأنواع يضيف تكلفة قراءة إضافية. الأفضل تعريف Schema صريح عندما يكون هيكل البيانات معروفاً.

قراءة الملفات أسرع عبر مخطط صريح

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

schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product", StringType(), True),
    StructField("category", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("price", DoubleType(), True),
    StructField("country", StringType(), True)
])

df = (
    spark.read
    .option("header", True)
    .schema(schema)
    .csv("data/large_sales_file.csv")
)

استكشاف البيانات دون تحميلها كاملة في الذاكرة

الفرق الجوهري بين PySpark وبين ما تعلمته سابقاً في مكتبة Pandas (2): استكشاف هيكل البيانات وفهم DataFrame و Series هو أن العمليات هنا لا تعني نسخ الجدول كاملاً إلى الذاكرة المحلية. بدلاً من ذلك، يتم إنشاء خطة تنفيذ ثم تشغيلها فقط عند الحاجة الفعلية.

يمكنك تنفيذ عمليات فحص أولية بسرعة، مثل عد السجلات، استخراج الأعمدة، أو الحصول على إحصاءات وصفية مبدئية. هذا مفيد جداً قبل البدء في التنظيف أو التحويل، تماماً كما ناقشنا في تنظيف البيانات (Data Cleaning): اكتشاف ومعالجة القيم المفقودة (Missing Values) لكن على نطاق أكبر بكثير.

df.select("product", "price", "quantity").show(10)

df.describe(["price", "quantity"]).show()

df.filter(df.price.isNull() | df.quantity.isNull()).count()

التصفية والتجميع السريع على ملايين السجلات

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

يمكن تطبيق مفاهيم مشابهة لما شرحناه في الفلترة المتقدمة (Filtering & Sorting): استخراج رؤى دقيقة من ملايين السجلات والتجميع والتلخيص (Groupby & Aggregation): إنشاء تقارير إحصائية برمجية، لكن مع محرك قادر على التعامل مع أحجام أكبر بكثير.

from pyspark.sql.functions import col, sum, avg, desc

filtered_df = df.filter(
    (col("country") == "Saudi Arabia") & (col("quantity") > 2)
)

report_df = (
    filtered_df.groupBy("category")
    .agg(
        sum(col("quantity")).alias("total_quantity"),
        avg(col("price")).alias("avg_price")
    )
    .orderBy(desc("total_quantity"))
)

report_df.show(truncate=False)

تنفيذ استعلامات SQL مباشرة

df.createOrReplaceTempView("sales")

result_sql = spark.sql("""
    SELECT
        category,
        COUNT(*) AS total_rows,
        SUM(quantity) AS total_quantity,
        ROUND(AVG(price), 2) AS avg_price
    FROM sales
    WHERE country = 'Saudi Arabia'
    GROUP BY category
    ORDER BY total_quantity DESC
""")

result_sql.show()

كيف تجعل القراءة والتحليل يتمان في ثوانٍ فعلاً؟

الوعد بالسرعة لا يتحقق تلقائياً بمجرد استخدام Spark. هناك مجموعة قرارات تقنية تصنع الفارق بين مهمة تستغرق ثوانٍ وأخرى تستنزف النظام دقائق طويلة.

• استخدم تنسيق Parquet بدلاً من CSV كلما أمكن، لأنه عمودي ومضغوط ويدعم قراءة الأعمدة المطلوبة فقط.
• حدّد الأعمدة الضرورية مبكراً بدلاً من قراءة كل شيء ثم التصفية لاحقاً.
• تجنب كثرة عمليات shuffle لأنها مكلفة شبكياً وقرصياً.
• استخدم cache() فقط عند الحاجة الفعلية لإعادة استخدام نفس النتائج عدة مرات.
• راقب خطة التنفيذ عبر explain() لفهم ما يجري فعلياً داخل المحرك.

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

مثال عملي: تنظيف وتحويل وتخزين النتائج

في بيئات الإنتاج، غالباً ما تكون القراءة مجرد مرحلة ضمن خط أكبر من بناء خطوط أنابيب البيانات (ETL – Extract, Transform, Load) باستخدام بايثون. بعد القراءة، تأتي عمليات تنظيف الصفوف الخاطئة، إنشاء أعمدة مشتقة، ثم تحميل النتائج إلى طبقة تحليلية جاهزة.

from pyspark.sql.functions import when

clean_df = (
    df.filter(col("price").isNotNull() & col("quantity").isNotNull())
      .filter((col("price") > 0) & (col("quantity") > 0))
      .withColumn("revenue", col("price") * col("quantity"))
      .withColumn(
          "sales_segment",
          when(col("revenue") >= 1000, "high")
          .when(col("revenue") >= 300, "medium")
          .otherwise("low")
      )
)

clean_df.write.mode("overwrite").parquet("output/sales_cleaned_parquet")

هذا الأسلوب يجعل البيانات جاهزة لاحقاً للتغذية في لوحات المعلومات، أو النماذج التنبؤية، أو التحليلات المتقدمة المرتبطة بمراحل مثل هندسة الميزات (Feature Engineering): كيف تستخرج بيانات جديدة من البيانات الحالية؟، بل وحتى تجهيزها لسيناريوهات مقدمة في تعلم الآلة (Machine Learning): الفرق بين التعلم الخاضع وغير الخاضع للإشراف.

متى تستخدم PySpark DataFrames فعلاً؟

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

أما إذا كانت البيانات صغيرة نسبياً، فقد تكون أدوات مثل Pandas أكثر بساطة وسرعة في التطوير. الفكرة الهندسية الصحيحة ليست استخدام أداة “أقوى” دائماً، بل استخدام الأداة المناسبة لحجم البيانات وتعقيد المعالجة.

الخلاصة

قراءة ملف بحجم جيجابايت في ثوانٍ ليست خدعة تقنية، بل نتيجة مباشرة لاستخدام محرك موزع مثل Apache Spark عبر واجهة PySpark DataFrames. السر الحقيقي يكمن في فهم بنية التنفيذ، وتقليل عمليات القراءة غير الضرورية، واختيار تنسيقات تخزين فعالة، ثم بناء خطوات تحليلية ذكية قابلة للتوسع.

كلما تحسنت قراراتك في تصميم البيانات، من المخطط الصريح إلى التقسيم والتخزين العمودي، اقتربت أكثر من أنظمة تحليل قادرة على تحويل الملفات الضخمة من عبء تقني إلى مصدر رؤى سريع وموثوق.