• ما نوع الوحش هذا تدفق الهواء ، وما المكونات التي يتكون منها وكيف يتفاعلون مع بعضهم البعض
• حول الكيانات الرئيسية لـ Airflow: خطوط الأنابيب المسماة DAG ، المشغل وعدد قليل من الأشياء الأخرى
• كيف تنجح في تطوير Airflow
• كيف نفذنا توليد خطوط الأنابيب وما يسمى "الكتابة التصريحية لخطوط الأنابيب"
• حول إيجابيات وسلبيات استخدام Airflow

ما هو تدفق الهواء

بنية تدفق الهواء

مكونات تدفق الهواء

• التحذير الأول هو أنه يمكن تشغيل مثيل جدولة واحد فقط في كل مرة. هذا يعني أن وضع High Availability غير ممكن حاليًا (يخطط المطورون لإضافة Scheduler HA إلى Airflow الإصدار 2.0).
• : , - . , - , .

• الأول ، الأبسط ، هو SequentialExecutor. يقوم بتشغيل المهام الواردة بالتسلسل ، وإيقاف المجدول مؤقتًا لمدة تنفيذها.
• LocalExecutor , , LocalExecutor . : - SQLite, LocalExecutor SequentialExecutor.
• CeleryExecutor , . Celery – , RabbitMQ Redis. , .
• DaskExecutor Dask – .
• KubernetesExecutor pod Kubernetes.
• DebugExecutor IDE.

كيانات Apache Airflow

dag = DAG(
   dag_id="load_some_data",
   schedule_interval="0 1 * * *",
   default_args={
       "start_date": datetime(2020, 4, 20),
       "owner": "DE",
       "depends_on_past": False,
       "sla": timedelta(minutes=45),
       "email": "<your_email_here>",
       "email_on_failure": True,
       "retries": 2,
       "retry_delay": timedelta(minutes=5)
   }
)

نقطة مهمة للغاية: نظرًا لأن Airflow تم تطويره بواسطة شركة دولية ، فإنه يعمل فقط في UTC. في الوقت الحالي ، لا توجد طريقة عقلانية لجعل Airflow يعمل في منطقة زمنية مختلفة ، لذلك عليك أن تتذكر باستمرار الفرق بين منطقتنا الزمنية والتوقيت العالمي المنسق (UTC). في الإصدار 1.10.10 ، أصبح من الممكن تغيير المنطقة الزمنية في واجهة المستخدم ، ولكن هذا ينطبق فقط على واجهة الويب ، وستظل خطوط الأنابيب تعمل بالتوقيت العالمي المنسق.

المشغل أو العامل

run_sql = HiveOperator(
   dag=dag,
   task_id="run_sql",
   hive_cli_conn_id="hive",
   hql="""
       INSERT OVERWRITE TABLE some_table
       SELECT * FROM other_table t1
       JOIN another_table t2 on ...
       WHERE other_table.dt = '{{ ds }}'
   """
)

notify = SlackAPIPostOperator(
   dag=dag,
   task_id="notify_slack",
   slack_conn_id="slack",
   token=token,
   channel="airflow_alerts",
   text="Guys, I'm done for {{ ds }}"
)

run_sql >> notify

العاطفة

مشغلي تدفق الهواء الأكثر شيوعًا

• BashOpetator و PythonOperator. كل شيء واضح معهم: يرسلون أمر bash ووظيفة python للتنفيذ ، على التوالي.
• هناك مجموعة كبيرة ومتنوعة من المشغلين لتقديم الاستفسارات إلى قواعد البيانات المختلفة. يتم دعم معايير Postgres و MySQL و Oracle و Hive و Presto. إذا لم يكن هناك عامل لقاعدة البيانات المفضلة لديك لسبب ما ، فيمكنك استخدام JdbcOperator أكثر عمومية أو كتابة الخاصة بك ، Airflow يسمح بذلك.
• Sensor – , . , - . , , . , : 3 , . . , , .
• BranchPythonOperator – , , python , , .
• DockerOpetator Docker- . , Docker- , . , .
• KubernetesPodOperator pod Kubernetes.
• DummyOperator , .

Lamoda

• – LamodaDockerOperator. , : - Hadoop, . LamodaDockerOperator Spark- , python.
• LamodaHiveperator – , . Hive. , - , , . , , HiveCliHook HiveServer2Hook, .
• – ExternalTaskSensor. . , Hadoop . , , , - , , . , - HDFS, Airflow.
• BashOperator, PythonOperator – , bash- python .
• , . - , .

Airflow

• Variables – , , , . , . , Hive, HDFS, . dev- prod-, .
• Connections – , . Airflow : http ftp, .
• Hooks – , .
• SLA -. , . SLA , , - - . - : - , Airflow .
• – XCom, cross-communication. : , json-. – 48 .
• – , . , . , 5, , , , .

كيف تنجح في تطوير تدفق الهواء

• من المفيد الاحتفاظ بكل خط أنابيب (أو مولد خط أنابيب ، أكثر من ذلك أدناه) في ملف منفصل. أعرف على الفور الملف الذي أحتاج إلى الانتقال إليه لإلقاء نظرة على خط الأنابيب أو المولد المطلوب.
• , , . , -, . , - , . : , , .
• – schedule_interval start_date dag_id. , Airflow , - -. DAGS , Scheduler, . , , dag_id. , .
• catchup. True, Airflow , start_date . , . False Airflow . , Airflow True ( -).
• – . , python , airflow DAG, , DAG. . , , . REST API, requests.get() .

:

from libs.dag_from_config.dag_generator import DagGenerator
from libs.runners.docker_runner import DockerRunner

generator = DagGenerator(config_dir='dag_configs/docker_runner', prefix='docker')
dags = generator.generate(task_runner=DockerRunner)

for dag in dags:
   globals()[dag.dag_id] = dag  #     

docker_image = "docker_registry/attribution/calculation:1.1.0"

dags {
 attribution {
   owner = "RND"
   name = "attribution"
   start_date = "20190601"
   emails = [...]
   schedule_interval = "0 1 * * *"
   depends_on_past = true
   concurrency = 4

   description = """
   -    z_log
   -        
   -  ,    
   -     
   """

   tags = ["critical"]

docker {
 connections {
   LMD_DWH = "dwh"
 }

 containers {
   desktop {
     image = ${docker_image}
     connections = [LMD_DWH]

     environment {
       LMD_YARN_QUEUE = "{{ var.value.YARN_QUEUE }}"
       LMD_INSTANCES = 60
       LMD_MEMORY_PER_INSTANCE = "4g"
       LMD_ZLOG_SOURCE = "z_log_db.z_log"
       LMD_ATTRIBUTION_TABLE = "{{ var.value.HIVE_DB_DERIVATIVES }}.z_log_attribution"
       LMD_ORDERS_TABLE = "rocket_dwh_bl.fct_orderitem_detail"
       LMD_PLATFORMS = "desktop"

       LMD_RUN_DATE = "{{ ds_nodash }}"
     }
   }
   mobile {...}
   iOS {...}
   Android {...}
 }
 tasks = [[desktop, mobile, iOS, Android]]
}

  awaits {
     z_log_compaction {
       dag = "compactor_daily_23_21_A_A_A"
       task = "compact_z_log_db_z_log"
       timedelta = 3hr37m
     }
     oracle_bl_fct_orderitem_detail {
       dag = "await_bl_fct_orderitem_detail_0_1_A_A_A"
     }
   }
 }
}

docker_image = "docker_registry/attribution/calculation:1.1.0"

dags {
 attribution {
   owner = "RND"
   name = "attribution"
   start_date = "20190601"
   emails = [...]
   schedule_interval = "0 1 * * *"
   depends_on_past = true
   concurrency = 4

   description = """
   -    z_log
   -        
   -  ,    
   -     
   """

   tags = ["critical"]


   docker {
     connections {
       LMD_DWH = "dwh"
     }

     containers {
       desktop {
         image = ${docker_image}
         connections = [LMD_DWH]

         environment {
           LMD_YARN_QUEUE = "{{ var.value.YARN_QUEUE }}"
           LMD_INSTANCES = 60
           LMD_MEMORY_PER_INSTANCE = "4g"
           LMD_ZLOG_SOURCE = "z_log_db.z_log"
           LMD_ATTRIBUTION_TABLE = "{{ var.value.HIVE_DB_DERIVATIVES }}.z_log_attribution"
           LMD_ORDERS_TABLE = "rocket_dwh_bl.fct_orderitem_detail"
           LMD_PLATFORMS = "desktop"

           LMD_RUN_DATE = "{{ ds_nodash }}"
         }
       }
       mobile {...}
       iOS {...}
       Android {...}
     }
     tasks = [[desktop, mobile, iOS, Android]]
   }


   awaits {
     z_log_compaction {
       dag = "compactor_daily_23_21_A_A_A"
       task = "compact_z_log_db_z_log"
       timedelta = 3hr37m
     }
     oracle_bl_fct_orderitem_detail {
       dag = "await_bl_fct_orderitem_detail_0_1_A_A_A"
     }
   }
 }
}

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

ماذا نريد أن نفعل بعد ذلك

لماذا اخترنا Airflow

1. أولاً ، هذا هو Python ، حيث بمساعدة حلقتين واثنين من الشروط ، يمكنك إنشاء خط أنابيب أنيق يعمل بشكل صحيح. ولن تحتاج إلى وصفها في جزء كبير من XML. بالإضافة إلى ذلك ، يتوفر نظام Python البيئي بالكامل تقريبًا وحديقة حيوانات المكتبات بأكملها خارج الصندوق ، والتي يمكن استخدامها كما تريد.
2. يبسط غياب XML مراجعة التعليمات البرمجية بشكل كبير. كتبنا رمز خط الأنابيب والتكوينات له ، وكل شيء رائع ، كل شيء يعمل. في الواقع ، يمكنك السحب بتنسيق XML أو أي تنسيق تكوين آخر ، ولكن هذه بالفعل مسألة ذوق.
3. unit-, , .
4. , «», . Airflow . , , .
5. Airflow ( ).
6. Active Directory RBAC (role-based access control, )
7. Worker Celery Kubernetes.
8. open source-, , .
9. Airflow , . .
10. : statsd , Sentry – , Airflow , . Airflow-exporter Prometheus.

Airflow,

1. – : , , execution_date – , .
2. - -, , , Apache NiFi. – code-review diff- , .
3. Scheduler - .
4. – , . – .
5. Airflow : . , , . RBAC ( ) , UI (, , ). RBAC – security Flask, .
6. : , , -, , . , .

Airflow

• crontab’a cron .
• Python.
• - Docker, , .
• , , real time.
• Airflow , “, , , Z – ”.

Airflow

• –
• Astronomer, hosted- Airflow Kubernetes. –
• Astronomer Airflow –
• Airflow () Slack ().