• quel genre de bête est ce flux d'air, de quels composants il se compose et comment ils interagissent les uns avec les autres
• à propos des principales entités d'Airflow: des pipelines appelés DAG, Operator et quelques autres choses
• comment réussir son développement sur Airflow
• comment nous avons implémenté la génération de pipelines et la soi-disant «écriture déclarative de pipelines»
• sur les avantages et les inconvénients de l'utilisation d'Airflow

Qu'est-ce que Airflow

Architecture du flux d'air

Composants de flux d'air

• La première mise en garde est qu'une seule instance de Scheduler peut être exécutée à la fois. Cela signifie que le mode Haute disponibilité n'est actuellement pas possible (les développeurs prévoient d'ajouter Scheduler HA à Airflow version 2.0).
• : , - . , - , .

• Le premier, le plus simple, est le SequentialExecutor. Il lance séquentiellement les tâches entrantes et met en pause le planificateur pendant la durée de leur exécution.
• LocalExecutor , , LocalExecutor . : - SQLite, LocalExecutor SequentialExecutor.
• CeleryExecutor , . Celery – , RabbitMQ Redis. , .
• DaskExecutor Dask – .
• KubernetesExecutor pod Kubernetes.
• DebugExecutor IDE.

Entités 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)
   }
)

Point très important: comme Airflow a été développé par une société internationale, il ne fonctionne qu'en UTC. Pour le moment, il n'y a pas de moyen sensé de faire fonctionner Airflow dans un fuseau horaire différent, vous devez donc constamment vous souvenir de la différence entre notre fuseau horaire et UTC. Dans la version 1.10.10, il est devenu possible de changer le fuseau horaire dans l'interface utilisateur, mais cela ne s'applique qu'à l'interface Web, les pipelines fonctionneront toujours en UTC.

Opérateur

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

Idempotence

Opérateurs Airflow les plus courants

• BashOpetator et PythonOperator. Tout est clair avec eux: ils envoient respectivement une commande bash et une fonction python pour exécution.
• Il existe une grande variété d'opérateurs pour envoyer des requêtes à diverses bases de données. Postgres standard, MySQL, Oracle, Hive, Presto sont pris en charge. Si pour une raison quelconque il n'y a pas d'opérateur pour votre base de données préférée, vous pouvez utiliser un JdbcOperator plus général ou écrire le vôtre, Airflow le permet.
• 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, , , , .

Comment réussir le développement du flux d'air

• Il est utile de conserver chaque pipeline (ou générateur de pipeline, plus d'informations ci-dessous) dans un fichier séparé. Je sais immédiatement dans quel fichier je dois aller pour voir le pipeline ou le générateur requis.
• , , . , -, . , - , . : , , .
• – 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"
     }
   }
 }
}

• 2 points dans le bloc d'attente transformés en deux capteurs qui attendent l'exécution du pipeline de base,
• Les 4 tâches que nous avons spécifiées dans le bloc docker se sont transformées en 4 tâches en cours d'exécution en parallèle,
• nous avons ajouté un DummyOperator entre les deux blocs d'opérateurs afin qu'il n'y ait pas de réseau de connexions entre les tâches.

Que voulons-nous faire ensuite

Pourquoi nous avons choisi Airflow

1. Tout d'abord, il s'agit de Python, où, à l'aide de deux boucles et de quelques conditions, vous pouvez créer un pipeline élégant et fonctionnant correctement. Et il n'aura pas besoin d'être décrit dans un énorme morceau de XML. De plus, presque tout l'écosystème Python et tout son zoo de bibliothèques sont disponibles prêts à l'emploi et peuvent être utilisés à votre guise.
2. L'absence de XML simplifie grandement la révision du code. Nous avons écrit le code et les configurations du pipeline, et tout est parfait, tout fonctionne. En fait, vous pouvez faire glisser en XML ou tout autre format de configuration, mais c'est déjà une question de goût.
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 ().