פיתוח מנוע זרימת עבודה של Apache Airflow במפתח מלא

נפח הנתונים שלך גדל, אבל ביצוע סקריפטים ידניים ומשימות cron כבר לא עומדים בקצב? נתקלנו במצבים שבהם שאילתות N+1 בצינור ETL גרמו לעיכובים של שעות, ללא ניטור במקום. צינור שטוען 5 מיליון הזמנות על LocalExecutor לקח 4 שעות, ובזמן עומס שיא, זמן השבתה הגיע ל-30 דקות. אנחנו פותרים בעיה זו עם Apache Airflow—פלטפורמה בוגרת לתזמור ואוטומציה של זרימות עבודה לעיבוד נתונים. המהנדסים שלנו בנו עשרות צינורות DAG לחברות במסחר אלקטרוני, פינטק ולוגיסטיקה. כל צינור עובר בדיקות עומס: DAG טיפוסי מעבד עד 10 מיליון רשומות בלילה, ועומד ב-SLA עם 99.9% מזמן הביצוע. היציבות מאושרת על ידי 50+ פרויקטים. קבלו ייעוץ לפרויקט שלכם—נעריך אותו תוך יום אחד.

Airflow לעומת Temporal ו-Camunda

Airflow מותאם לעיבוד נתונים אצוות:

• צינורות ETL/ELT (PostgreSQL → טרנספורמציה → מחסן נתונים)
• דוחות וייצוא יומיים
• צינורות ML (הכנת נתונים → אימון → פריסת מודל)
• אגרגציות וסנכרונים תקופתיים

אם יש לך תהליכים עסקיים מונעי אירועים עם משימות אנושיות, שקול Temporal או Camunda. Airflow לא יכול לחכות למשתמשים במשך שעות. אבל עבור מהנדסי נתונים, זו הבחירה הטובה ביותר: הוא מהיר פי 3–5 מ-Temporal בתרחישי אצווה.

למה אנחנו בוחרים ב-KubernetesExecutor

KubernetesExecutor מספק בידוד ברמת המשימה: כל משימה רצה ב-Pod משלה עם CPU וזיכרון ייעודיים. בעומס שיא, Kubernetes מפעיל אוטומטית Pods ומקטין אותם לאחר מכן. אנחנו משתמשים בגישה זו בייצור ורואים בה את התקן לצינורות נתונים מודרניים.

איך KubernetesExecutor מאיץ עיבוד נתונים

בעומס שיא, KubernetesExecutor מרחיב משאבים אופקית: במקום צומת אחד, נעשה שימוש באשכול של 10+ Pods. זמן הביצוע של צינור ETL מופחת פי 3–5 בהשוואה ל-LocalExecutor. לדוגמה, צינור שטוען 5 מיליון הזמנות על LocalExecutor לוקח 4 שעות; על KubernetesExecutor, זה לוקח 50 דקות. ההבדל ברור.

איך אנחנו בונים צינור ETL: מקרה מבחן מפורט

שקול דוגמה לחנות מקוונת: טעינה יומית של הזמנות מ-PostgreSQL ל-DWH (PostgreSQL) עם טרנספורמציה ואגרגציה. להלן DAG שפרסנו עבור לקוח תוך יומיים.

התקנה דרך Helm

helm repo add apache-airflow https://airflow.apache.org
helm upgrade --install airflow apache-airflow/airflow \
  --namespace airflow \
  --create-namespace \
  --set executor=KubernetesExecutor \
  --set postgresql.enabled=true \
  --set redis.enabled=true \
  --values airflow-values.yaml

---
# airflow-values.yaml aiflow:
image:
  repository: apache/airflow
  tag: 2.8.0
config:
  AIRFLOW__CORE__DAGS_FOLDER: /opt/airflow/dags
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: "3"
  AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL: "30"
dags:
  gitSync:
    enabled: true
    repo: https://github.com/company/airflow-dags.git
    branch: main
    subPath: dags/
---

DAG — דוגמה לצינור ETL

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta
import pandas as pd

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['[email protected]'],
}

with DAG(
    'daily_orders_etl',
    default_args=default_args,
    schedule_interval='0 2 * * *',
    catchup=False,
    tags=['etl', 'orders'],
    description='Загрузка и трансформация заказов в DWH',
) as dag:

    def extract_orders(**context):
        hook = PostgresHook(postgres_conn_id='production_db')
        ds = context['ds']
        df = hook.get_pandas_df(f"""
            SELECT o.id, o.customer_id, o.total, o.status, o.created_at,
                   c.email, c.country
            FROM orders o
            JOIN customers c ON c.id = o.customer_id
            WHERE o.created_at::date = '{ds}'
              AND o.status IN ('paid', 'shipped', 'delivered')
        """)
        context['ti'].xcom_push(key='orders_count', value=len(df))
        df.to_parquet(f'/tmp/orders_{ds}.parquet')
        return len(df)

    def transform_orders(**context):
        ds = context['ds']
        df = pd.read_parquet(f'/tmp/orders_{ds}.parquet')
        df['order_date'] = pd.to_datetime(df['created_at']).dt.date
        df['revenue_usd'] = df['total'] / 100
        df['is_international'] = df['country'] != 'RU'
        df['customer_tier'] = df['revenue_usd'].apply(
            lambda x: 'vip' if x >= 500 else 'regular'
        )
        df.to_parquet(f'/tmp/orders_transformed_{ds}.parquet')

    def load_to_dwh(**context):
        ds = context['ds']
        df = pd.read_parquet(f'/tmp/orders_transformed_{ds}.parquet')
        hook = PostgresHook(postgres_conn_id='datawarehouse')
        engine = hook.get_sqlalchemy_engine()
        df.to_sql('fact_orders', engine, schema='dwh', if_exists='append',
                  index=False, method='multi', chunksize=1000)

    aggregate_metrics = PostgresOperator(
        task_id='aggregate_metrics',
        postgres_conn_id='datawarehouse',
        sql="""
            INSERT INTO dwh.daily_metrics (date, total_revenue, orders_count, avg_order)
            SELECT '{{ ds }}'::date, SUM(revenue_usd), COUNT(*), AVG(revenue_usd)
            FROM dwh.fact_orders
            WHERE order_date = '{{ ds }}'
            ON CONFLICT (date) DO UPDATE SET
                total_revenue = EXCLUDED.total_revenue,
                orders_count = EXCLUDED.orders_count,
                avg_order = EXCLUDED.avg_order;
        """,
    )

    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders
    )
    transform = PythonOperator(
        task_id='transform_orders',
        python_callable=transform_orders
    )
    load = PythonOperator(
        task_id='load_to_dwh',
        python_callable=load_to_dwh
    )

    extract >> transform >> load >> aggregate_metrics

ביצוע מקביל ו-Sensors

from airflow.utils.task_group import TaskGroup

with TaskGroup('process_regions') as process_regions:
    for region in ['EU', 'US', 'APAC']:
        PythonOperator(
            task_id=f'process_{region.lower()}',
            python_callable=process_region_data,
            op_kwargs={'region': region}
        )

extract >> process_regions >> aggregate_all

כדי לחכות לאירועים חיצוניים, אנו משתמשים ב-Sensors: helm repo add apache-airflow https://airflow.apache.org helm upgrade --install airflow apache-airflow/airflow \ --namespace airflow \ --create-namespace \ --set executor=KubernetesExecutor \ --set postgresql.enabled=true \ --set redis.enabled=true \ --values airflow-values.yaml לקבצים, # airflow-values.yaml aiflow: image: repository: apache/airflow tag: 2.8.0 config: AIRFLOW__CORE__DAGS_FOLDER: /opt/airflow/dags AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: "3" AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL: "30" dags: gitSync: enabled: true repo: https://github.com/company/airflow-dags.git branch: main subPath: dags/ ל-APIs. זהו תבנית ייצור סטנדרטית.

KubernetesExecutor בפעולה

executor_config = {
  'KubernetesExecutor': {
    'request_memory': '2Gi',
    'request_cpu': '500m',
    'limit_memory': '4Gi',
    'image': 'custom-airflow:2.8.0-pandas',
  }
}
heavy_transform = PythonOperator(
  task_id='heavy_transform',
  python_callable=transform_large_dataset,
  executor_config=executor_config
)

איך מטפלים בכשלי DAG

כאשר משימה נכשלת, Airflow מנסה שוב אוטומטית עם backoff אקספוננציאלי. אנו מגדירים התראות ב-Telegram/Slack עבור כל כשל ומשימה ארוכת טווח. בנוסף, אנו משלבים מדדים ב-Prometheus/Grafana: לוחות מחוונים מציגים זמן ביצוע, מספר משימות מוצלחות/נכשלות, וניצול משאבים. זה מאפשר תגובה מהירה לאירועים.

דרישות תשתית מינימליות:

• אשכול Kubernetes בגרסה 1.24+
• PostgreSQL 13+ למסד נתונים מטא-דאטה
• Redis (אופציונלי, עבור CeleryExecutor)
• קיבולת אחסון: מ-100 GB עבור לוגים וארטיפקטים

סקירת תהליך

1. ניתוח: אנו לומדים את מקורות הנתונים שלך, נפחים, SLA, בעיות נוכחיות, תצורת Airflow ותשתית.
2. עיצוב: אנו מתארים DAGs, בוחרים Executor, מתכננים טיפול בשגיאות והתראות.
3. יישום: אנו כותבים קוד DAG, תצורות Helm, CI/CD ובדיקות.
4. פריסה: אנו מקימים תשתית (Kubernetes או Docker), מגדירים GitSync וניטור.
5. בדיקות: אנו מריצים backfill על נתונים היסטוריים ומוודאים תקינות.
6. תיעוד והדרכה: אנו מספקים runbook ומכשירים את הצוות שלך.
7. תמיכה: שבועיים לאחר ההשקה לייצוב.

מה כלול בעבודה

• DAGs שפותחו (קוד Python מוכן לייצור)
• תצורת Airflow (values.yaml, משתנים, חיבורים)
• צינור CI/CD לפריסה אוטומטית של DAG דרך Git
• תמונת Docker עם תלויות (ספריות, דרייברים)
• תיעוד: תיאורי DAG, הוראות הרצה, פתרון בעיות
• הדרכת צוות (2–3 מפגשים)
• ניטור: התראות ב-Telegram/Slack, לוחות מחוונים של Grafana

הערכות לוחות זמנים

הניסיון והערבויות שלנו

יש לנו 10+ שנות ניסיון בפיתוח תשתית נתונים. סיפקנו 50+ פרויקטי Airflow למסחר אלקטרוני, פינטק ולוגיסטיקה. אנו משתמשים בשיטות עבודה מומלצות: GitSync, KubernetesExecutor, התראות, גרסאות DAG. אנו מבטיחים פעילות יציבה של הצינור ותיעוד ברמה ארגונית.

אנחנו מוכנים לבנות צינור דומה עבור הנתונים שלך. צור קשר להערכה—נגיב תוך יום אחד. או בקש ייעוץ לדיון בפרטים ללא התחייבות.

תיעוד Apache Airflow