Technical Tutorials

مع انفجار أحجام البيانات وتوزيع البنية التحتية بشكل متزايد، لم تكن الحاجة إلى تنسيق سير عمل موثوق وقابل للمراقبة والتوسعة أكبر من أي وقت مضى. برز Apache Airflow كمعيار صناعي فعلي لهذا التحدي. من خلال التعبير عن سير العمل كمخططات دائرية غير دورية (DAGs) باستخدام كود Python، يتيح Airflow لمهندسي البيانات الانتقال بعيداً عن سكربتات shell الهشة نحو خطوط بيانات قابلة للتحكم بالإصدارات، والمختبرة، والقابلة للصيانة.

سواء كنت تؤتمت مهام الدفعة الليلية البسيطة أو تنسق خدمات المايكروسيرفس المعقدة في الوقت الفعلي، فإن فهم المكونات الأساسية لـ Airflow—DAGs والمشغلات والمستشعرات واستراتيجيات التنفيذ—ضروري لأي مجموعة بيانات حديثة. في هذا الدليل، سنحلل البنية المعمارية، ونستكشف أمثلة كود عملية، ونناقش الاعتبارات الحرجة للنشر في بيئة الإنتاج.

التجريد الأساسي: DAGs والجدولة

في قلب Airflow يوجد DAG (المخطط الدائري غير الدوري). DAG هو مجموعة من المهام ذات الاعتماديات، ممثلة كعقد (مهام) وحواف (اعتماديات). على عكس أنظمة الجدولة التقليدية التي تعتمد على ملفات إعدادات ثابتة، يعرّف Airflow الـ DAGs في Python. هذا يتيح التوليد الديناميكي للمهام، والمنطق الشرطي، وتخصيص المعلمات مباشرة داخل قاعدة الكود الخاصة بك.

عند نشر DAG إلى واجهة Airflow، يفحص محرك الجدولة مجلد ~/.airflow/dags. يقوم بتحليل ملف Python، ويحدد الجدول الزمني (على سبيل المثال، بنمط cron)، وينشئ "تشغيلات DAG" لكل فترة مجدولة. فهم معلمات start_date وschedule_interval أمر حاسم، لأنها تحدد السياق الزمني لخط البيانات الخاص بك.

لبنات البناء: المشغلات والمستشعرات والخطافات

تُنفذ المهام في Airflow بواسطة المشغلات (Operators). المشغل هو فئة Python تعرّف إجراءً أو خطوة واحدة في الخط. يوفر Airflow نظاماً بيئياً ضخماً من المشغلات الجاهزة للخدمات مثل AWS وGCP وDBT وKubernetes وSQL. ومع ذلك، عندما تكون هناك حاجة إلى منطق محدد، يمكنك كتابة مشغلات مخصصة أو استخدام PythonOperator.

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

توفر الخطافات (Hooks) واجهة منخفضة المستوى للخدمات الخارجية، وتتعامل مع المصادقة واستدعاءات API، ثم تُغلّف بواسطة المشغلات والمستشعرات لتجريد على مستوى أعلى.

مثال عملي: خط ETL

فكّر في سيناريو ETL شائع: استخراج البيانات من قاعدة بيانات PostgreSQL، وتحويلها باستخدام Pandas، وتحميلها في بحيرة بيانات S3.

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.apache.hive.operators.hive import HiveOperator
from airflow.operators.python import PythonOperator
from airflow.providers.aws.operators.s3 import S3Operator
from airflow.sensors.postgres import PostgresSensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

def extract_custom_logic(**context):
    # Custom Python logic for transformation
    data = context['task_instance'].xcom_pull(task_ids='extract_sql')
    processed_data = transform(data) 
    return processed_data

def load_to_s3(processed_data):
    # Logic to upload to S3
    pass

with DAG(
    dag_id='etl_sales_pipeline',
    default_args=default_args,
    description='ETL pipeline for sales data',
    schedule_interval='@daily',
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=['etl', 'sales'],
) as dag:

    wait_for_data = PostgresSensor(
        task_id='wait_for_new_data',
        sql="SELECT 1 FROM sales WHERE updated_at > now() - interval '1 day'",
        postgres_conn_id='prod_postgres',
        poke_interval=60,
    )

    extract = PostgresOperator(
        task_id='extract_sql',
        sql="SELECT * FROM sales WHERE updated_at > now() - interval '1 day'",
        postgres_conn_id='prod_postgres',
    )

    transform = PythonOperator(
        task_id='transform_data',
        python_callable=extract_custom_logic,
    )

    load = S3Operator(
        task_id='load_to_s3',
        s3_key='data/sales/etl_output.parquet',
        local_location='/tmp/sales_output.parquet',
    )

    wait_for_data >> extract >> transform >> load

في هذا المثال، نعرّف خطاً قوياً مع إعادة المحاولة ومستشعر لضمان توفر البيانات. يتيح xcom_pull للمشغل Python استلام البيانات من مهمة SQL السابقة، مما يوضح كيف يتم تمرير الحالة بين الخطوات.

أفضل الممارسات للنشر في بيئة الإنتاج

نقل Airflow إلى بيئة الإنتاج يتطلب أكثر من مجرد تثبيت البرنامج. تشمل أفضل الممارسات الرئيسية:

فصل الاهتمامات: حافظ على تعريفات DAG في Git، منفصلة عن وحدات عامل Airflow.
اختيار المنفذ (Executor): استخدم منفذ Celery أو Kubernetes للتوسع الأفقي، وتجنب منفذ العملية الواحدة (Single Process) لأحمال الإنتاج.
إدارة الاتصالات: خزّن الأسرار في متغيرات البيئة أو مدير أسرار (مثل AWS Secrets Manager) وارجع إليها عبر connections بدلاً من ترميز بيانات الاعتماد يدوياً.
المراقبة: قم بدمج Airflow مع أدوات المراقبة مثل Prometheus وGrafana أو Datadog لتتبع مدة المهام ومعدلات الفشل واستخدام الموارد.
الاختبار: استخدم أطر عمل مثل pytest لاختبار منطق DAG بشكل وحداتي وairflow dags test لاختبارات التكامل قبل النشر.

الخاتمة

يوفر Apache Airflow إطاراً قوياً ومرناً وقابلاً للتوسعة لتنسيق سير عمل البيانات. من خلال الاستفادة من صياغة DAG التوضيحية، والنظام البيئي الغني للمشغلات، وقدرات الجدولة القوية، يمكن لفِرق هندسة البيانات بناء خطوط بيانات ليست فقط فعالة بل أيضاً مرنة وقابلة للصيانة. مع نمو بنية بياناتك التحتية، فإن الاستثمار في تصميم Airflow المناسب وتقوية بيئة الإنتاج سيحقق عوائد في الموثوقية والكفاءة التشغيلية.