استخراج البيانات (Extract): سحب ملايين السجلات من واجهات API وقواعد بيانات SQL

استخراج البيانات (Extract): سحب ملايين السجلات من واجهات API وقواعد بيانات SQL

تبدأ أي منظومة تحليلية أو تشغيلية قوية من مرحلة Extract، وهي الخطوة التي يتم فيها جلب البيانات الخام من مصادر متعددة قبل تنظيفها وتحويلها وتخزينها. في البيئات الصغيرة قد يبدو الأمر مجرد استدعاء ملف أو تنفيذ استعلام بسيط، لكن عند التعامل مع ملايين السجلات من واجهات API أو قواعد SQL يصبح الاستخراج تحدياً هندسياً يتطلب تصميماً واعياً وقابلاً للتوسع.

هذا المقال يشرح كيف تُبنى طبقة استخراج احترافية ضمن بناء خطوط أنابيب البيانات (ETL – Extract, Transform, Load) باستخدام بايثون، مع التركيز على الاعتمادية، السرعة، تقليل الضغط على المصدر، والحفاظ على جودة البيانات منذ أول لحظة. كما يرتبط هذا المفهوم مباشرة بما تم توضيحه في مقدمة في هندسة البيانات (Data Engineering): كيف تتعامل الشركات مع “البيانات الضخمة”؟.

لماذا تعد مرحلة الاستخراج أصعب مما تبدو عليه؟

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

عند سحب بيانات ضخمة، يجب التفكير في عناصر مثل pagination، والتجزئة الزمنية، والاسترجاع عند الفشل، وتحديد نقطة الاستئناف checkpoint، والتحكم في الذاكرة، وفصل السجلات غير الصالحة عن الصالحة. هذه التفاصيل هي ما يجعل خط الاستخراج موثوقاً على المدى الطويل.

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

استخراج البيانات من واجهات API على نطاق كبير

فهم طبيعة الواجهة قبل كتابة أي كود

قبل كتابة أول سطر برمجي، افحص توثيق الواجهة بعناية: هل تدعم offset pagination أم cursor pagination؟ هل يوجد حد أقصى للطلبات rate limit؟ هل يمكن فلترة البيانات حسب تاريخ آخر تعديل؟ هذه الأسئلة تحدد شكل بنية الاستخراج بالكامل.

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

مثال عملي لاستخراج تزايدي من API

import requests
import pandas as pd
import time

BASE_URL = "https://api.example.com/v1/orders"
API_KEY = "your_api_key"
LAST_SYNC = "2024-01-01T00:00:00Z"

headers = {"Authorization": f"Bearer {API_KEY}"}
params = {
    "updated_after": LAST_SYNC,
    "limit": 1000
}

all_rows = []
next_cursor = None

while True:
    if next_cursor:
        params["cursor"] = next_cursor

    response = requests.get(BASE_URL, headers=headers, params=params, timeout=30)
    response.raise_for_status()

    payload = response.json()
    rows = payload.get("data", [])
    all_rows.extend(rows)

    next_cursor = payload.get("next_cursor")

    if not next_cursor:
        break

    time.sleep(0.2)

df = pd.DataFrame(all_rows)
print(df.shape)
print(df.head())

في المثال السابق استخدمنا استخراجاً تزايدياً اعتماداً على الحقل updated_after. هذا النمط ينسجم لاحقاً مع مراحل الفحص والتحليل في مكتبة Pandas (1): قراءة واستدعاء البيانات من ملفات CSV و Excel برمجياً، ثم استكشاف البنية عبر مكتبة Pandas (2): استكشاف هيكل البيانات وفهم DataFrame و Series.

ممارسات ضرورية عند استخراج API

• استخدم إعادة المحاولة retry عند أخطاء الشبكة أو الاستجابات المؤقتة.
• احترم حدود rate limiting ولا ترسل طلبات عدوانية تؤدي إلى الحظر.
• سجل بيانات المراقبة مثل زمن الاستجابة، عدد الصفحات، ونسبة السجلات الناقصة.
• احفظ النسخة الخام raw layer قبل أي تحويل.

استخراج البيانات من قواعد SQL بكفاءة

الفرق بين الاستعلام التحليلي والاستعلام الاستخراجي

عند سحب ملايين الصفوف من قاعدة SQL، لا يكفي أن يكون الاستعلام “صحيحاً” منطقياً، بل يجب أن يكون اقتصادياً في استخدام الفهارس والذاكرة والشبكة. الاستعلام الذي يعمل جيداً على 50 ألف سجل قد ينهار تماماً عند 200 مليون سجل.

أفضل الممارسات تبدأ بقراءة الأعمدة المطلوبة فقط، وتجنب SELECT *، ثم استخدام شروط زمنية أو نطاقات معرفات لتقسيم القراءة إلى دفعات. كما يجب فهم خطة التنفيذ execution plan قبل تشغيل أي مهمة كبيرة.

مثال بايثون لاستخراج دفعات من قاعدة SQL

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2://user:password@host:5432/dbname")

query = """
SELECT order_id, customer_id, order_date, total_amount, updated_at
FROM sales_orders
WHERE updated_at >= '2024-01-01'
ORDER BY updated_at
"""

chunks = pd.read_sql(query, engine, chunksize=50000)

total_rows = 0

for i, chunk in enumerate(chunks, start=1):
    total_rows += len(chunk)
    output_file = f"orders_chunk_{i}.parquet"
    chunk.to_parquet(output_file, index=False)
    print(f"Saved {output_file} with {len(chunk)} rows")

print(f"Total extracted rows: {total_rows}")

ميزة هذا الأسلوب أنه يمنع تحميل كامل الجدول في الذاكرة. وبعد الاستخراج يمكن متابعة مراحل تنظيف البيانات (Data Cleaning): اكتشاف ومعالجة القيم المفقودة (Missing Values)، ومعالجة التشوهات عبر معالجة البيانات المكررة والمشوهة (Duplicates & Outliers) باستخدام بايثون.

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

متى نستخدم Apache Spark بدلاً من Pandas؟

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

هذا المستوى من العمل يرتبط بجوهر Big Data وبيئات Hadoop، حيث يكون التخزين والمعالجة منفصلين لكن متكاملين على مستوى البنية.

مثال PySpark لقراءة جدول كبير عبر JDBC

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ExtractLargeSQLTable") \
    .getOrCreate()

df = spark.read.format("jdbc") \
    .option("url", "jdbc:postgresql://host:5432/dbname") \
    .option("dbtable", "public.sales_orders") \
    .option("user", "user") \
    .option("password", "password") \
    .option("driver", "org.postgresql.Driver") \
    .option("partitionColumn", "order_id") \
    .option("lowerBound", 1) \
    .option("upperBound", 10000000) \
    .option("numPartitions", 16) \
    .load()

print(df.count())
df.write.mode("overwrite").parquet("/data/raw/sales_orders")

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

تصميم طبقة استخراج قابلة للاعتماد

لكي تنجح مرحلة Extract في الإنتاج، يجب أن تشمل أكثر من مجرد كود القراءة. هناك مكونات تشغيلية لا تقل أهمية عن المنطق البرمجي نفسه.

1. تعريف مصدر البيانات وحقول الاعتماد والمصادقة.
2. تحديد استراتيجية السحب: كامل full load أو تزايدي incremental.
3. حفظ البيانات الخام في طبقة مؤقتة يمكن الرجوع إليها.
4. إضافة سجلات مراقبة logs ومؤشرات جودة.
5. بناء آلية استئناف عند الفشل دون إعادة كل العمل من البداية.

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

أخطاء شائعة يجب تجنبها

• قراءة كل السجلات يومياً رغم أن 1% فقط منها تغيّر.
• سحب أعمدة غير مستخدمة ترفع كلفة التخزين والنقل.
• عدم اختبار الاستعلام على بيئة مماثلة للإنتاج.
• إهمال الترميز الزمني timezone في البيانات متعددة المناطق.
• تجاهل السجلات الفاشلة بدلاً من عزلها وتحليل سبب الفشل.

الخلاصة

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

سواء كنت تسحب من API أو من قواعد SQL أو عبر Spark، فإن النجاح الحقيقي يكمن في بناء طبقة استخراج مرنة، قابلة للمراقبة، وقادرة على النمو مع نمو الأعمال والبيانات.