• que tipo de besta é este Airflow, em quais componentes ele consiste e como eles interagem entre si
• sobre as principais entidades do Airflow: pipelines chamados DAG, Operator e mais algumas coisas
• como ter sucesso no desenvolvimento no Airflow
• como implementamos a geração de pipelines e a chamada "escrita declarativa de pipelines"
• sobre os prós e contras de usar o Airflow

O que é Airflow

Arquitetura Airflow

Componentes de fluxo de ar

• A primeira ressalva é que apenas uma instância do Scheduler pode estar em execução por vez. Isso significa que no momento é impossível operar em alta disponibilidade (os desenvolvedores estão planejando adicionar Scheduler HA ao Airflow versão 2.0).
• : , - . , - , .

• O primeiro, o mais simples, é o SequentialExecutor. Ele inicia tarefas de entrada sequencialmente e pausa o agendador durante sua execução.
• LocalExecutor , , LocalExecutor . : - SQLite, LocalExecutor SequentialExecutor.
• CeleryExecutor , . Celery – , RabbitMQ Redis. , .
• DaskExecutor Dask – .
• KubernetesExecutor pod Kubernetes.
• DebugExecutor IDE.

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

Ponto muito importante: como o Airflow foi desenvolvido por uma empresa internacional, ele só funciona na UTC. No momento, não existe uma maneira sensata de fazer o Airflow funcionar em um fuso horário diferente, então você precisa se lembrar constantemente da diferença entre nosso fuso horário e o UTC. Na versão 1.10.10, tornou-se possível alterar o fuso horário na IU, mas isso só se aplica à interface da web, os pipelines ainda serão executados em UTC.

Operador

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

Idempotência

Operadores mais comuns do Airflow

• BashOpetator e PythonOperator. Tudo é claro com eles: eles enviam um comando bash e uma função python para execução, respectivamente.
• Há uma grande variedade de operadores para enviar consultas a vários bancos de dados. Postgres padrão, MySQL, Oracle, Hive, Presto são suportados. Se por algum motivo não houver um operador para seu banco de dados favorito, você pode usar um JdbcOperator mais geral ou escrever o seu próprio, o Airflow permite isso.
• 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, , , , .

Como ter sucesso no desenvolvimento de fluxo de ar

• É útil manter cada pipeline (ou gerador de pipeline, mais sobre isso abaixo) em um arquivo separado. Eu sei imediatamente qual arquivo preciso acessar para examinar o pipeline ou gerador necessário.
• , , . , -, . , - , . : , , .
• – 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 pontos no bloco awaits transformados em dois sensores que aguardam a execução do pipeline básico,
• As 4 tarefas que especificamos no bloco docker se transformaram em 4 tarefas em execução em paralelo,
• adicionamos um DummyOperator entre os dois blocos de operadores para que não haja uma rede de conexões entre as tarefas.

O que queremos fazer a seguir

Por que escolhemos o Airflow

1. Em primeiro lugar, este é o Python, onde com a ajuda de dois loops e algumas condições, você pode fazer um pipeline elegante e funcionando corretamente. E não precisará ser descrito em um grande pedaço de XML. Além disso, quase todo o ecossistema Python e todo o zoológico de bibliotecas estão disponíveis prontos para uso, que podem ser usados ​​como você quiser.
2. A ausência de XML simplifica muito a revisão do código. Escrevemos o código do pipeline e as configurações para ele e está tudo ótimo, tudo funciona. Na verdade, você pode arrastar XML ou qualquer outro formato de configuração, mas isso já é uma questão de gosto.
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 ().