• Was für ein Biest ist dieser Luftstrom, aus welchen Komponenten er besteht und wie sie miteinander interagieren
• über die Hauptentitäten von Airflow: Pipelines namens DAG, Operator und ein paar weitere Dinge
• wie man bei Airflow erfolgreich entwickelt
• wie wir die Erzeugung von Pipelines und das sogenannte "deklarative Schreiben von Pipelines" implementiert haben
• über die Vor- und Nachteile der Verwendung von Airflow

Was ist Luftstrom?

Luftstromarchitektur

Luftstromkomponenten

• Die erste Einschränkung ist, dass jeweils nur eine Scheduler-Instanz ausgeführt werden kann. Dies bedeutet, dass es derzeit unmöglich ist, mit Hochverfügbarkeit zu arbeiten (die Entwickler planen, Scheduler HA zu Airflow Version 2.0 hinzuzufügen).
• : , - . , - , .

• Der erste, der einfachste ist der SequentialExecutor. Es startet nacheinander eingehende Aufgaben und hält den Scheduler für die Dauer ihrer Ausführung an.
• LocalExecutor , , LocalExecutor . : - SQLite, LocalExecutor SequentialExecutor.
• CeleryExecutor , . Celery – , RabbitMQ Redis. , .
• DaskExecutor Dask – .
• KubernetesExecutor pod Kubernetes.
• DebugExecutor IDE.

Apache Airflow-Entitäten

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)
   }
)

Sehr wichtiger Punkt: Da Airflow von einem internationalen Unternehmen entwickelt wurde, funktioniert es nur in UTC. Im Moment gibt es keine vernünftige Möglichkeit, Airflow in einer anderen Zeitzone zum Laufen zu bringen. Sie müssen sich daher ständig an den Unterschied zwischen unserer Zeitzone und UTC erinnern. In Version 1.10.10 wurde es möglich, die Zeitzone in der Benutzeroberfläche zu ändern. Dies gilt jedoch nur für die Weboberfläche. Pipelines werden weiterhin in UTC ausgeführt.

Operator

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

Idempotenz

Die gängigsten Luftstrombetreiber

• BashOpetator und PythonOperator. Mit ihnen ist alles klar: Sie senden einen Bash-Befehl bzw. eine Python-Funktion zur Ausführung.
• Es gibt eine Vielzahl von Operatoren zum Senden von Abfragen an verschiedene Datenbanken. Standard Postgres, MySQL, Oracle, Hive, Presto werden unterstützt. Wenn es aus irgendeinem Grund keinen Operator für Ihre Lieblingsdatenbank gibt, können Sie einen allgemeineren JdbcOperator verwenden oder Ihren eigenen schreiben, Airflow erlaubt dies.
• 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, , , , .

Erfolg in der Luftstromentwicklung

• Es ist nützlich, jede Pipeline (oder jeden Pipeline-Generator, mehr dazu weiter unten) in einer separaten Datei zu speichern. Ich weiß sofort, zu welcher Datei ich gehen muss, um die erforderliche Pipeline oder den erforderlichen Generator anzuzeigen.
• , , . , -, . , - , . : , , .
• – 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 Punkte im Warteblock wurden zu zwei Sensoren, die auf die Ausführung der Basis-Pipeline warten.
• Die 4 Aufgaben, die wir im Docker-Block angegeben haben, wurden zu 4 parallel laufenden Aufgaben.
• Wir haben einen DummyOperator zwischen den beiden Operatorblöcken hinzugefügt, damit kein Verbindungsnetz zwischen Aufgaben besteht.

Was wollen wir als nächstes tun?

Warum wir uns für Airflow entschieden haben

1. Erstens ist dies Python, wo Sie mit Hilfe von zwei Schleifen und einigen Bedingungen eine elegante, korrekt funktionierende Pipeline erstellen können. Und es muss nicht in einem großen Teil von XML beschrieben werden. Außerdem ist fast das gesamte Python-Ökosystem und sein gesamter Bibliothekszoo sofort verfügbar und können nach Belieben verwendet werden.
2. Das Fehlen von XML vereinfacht die Codeüberprüfung erheblich. Wir haben den Pipeline-Code und die Konfigurationen dafür geschrieben, und alles ist in Ordnung, alles funktioniert. Sie können zwar XML oder ein anderes Konfigurationsformat ziehen, aber das ist Geschmackssache.
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 ().