• Jenis binatang apa Aliran Udara ini, terdiri dari komponen apa dan bagaimana mereka berinteraksi satu sama lain
• tentang entitas utama Aliran Udara: saluran pipa yang disebut DAG, Operator, dan beberapa hal lainnya
• bagaimana berhasil dalam pengembangan di Airflow
• bagaimana kami menerapkan pembuatan pipelines dan apa yang disebut "penulisan deklaratif pipelines"
• tentang pro dan kontra menggunakan Airflow

Apa itu Airflow

Arsitektur aliran udara

Komponen aliran udara

• Peringatan pertama adalah bahwa hanya satu instance Scheduler yang dapat berjalan dalam satu waktu. Ini berarti bahwa saat ini tidak mungkin untuk beroperasi dalam Ketersediaan Tinggi (pengembang berencana menambahkan Scheduler HA ke Airflow versi 2.0).
• : , - . , - , .

• Yang pertama, yang paling sederhana, adalah SequentialExecutor. Ini secara berurutan meluncurkan tugas-tugas masuk, dan menjeda penjadwal selama pelaksanaannya.
• LocalExecutor , , LocalExecutor . : - SQLite, LocalExecutor SequentialExecutor.
• CeleryExecutor , . Celery – , RabbitMQ Redis. , .
• DaskExecutor Dask – .
• KubernetesExecutor pod Kubernetes.
• DebugExecutor IDE.

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

Poin yang sangat penting: karena Airflow dikembangkan oleh perusahaan internasional, Airflow hanya berfungsi di UTC. Saat ini, tidak ada cara yang masuk akal untuk membuat Aliran Udara berfungsi di zona waktu yang berbeda, jadi Anda harus terus mengingat tentang perbedaan antara zona waktu kita dan UTC. Di versi 1.10.10, dimungkinkan untuk mengubah zona waktu di UI, tetapi ini hanya berlaku untuk antarmuka web, pipeline akan tetap berjalan dalam UTC.

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

Idempotensi

Operator aliran udara yang paling umum

• BashOpetator dan PythonOperator. Semuanya jelas dengan mereka: mereka masing-masing mengirim perintah bash dan fungsi python untuk eksekusi.
• Ada berbagai macam operator untuk mengirimkan pertanyaan ke berbagai database. Postgres standar, MySQL, Oracle, Hive, Presto didukung. Jika karena alasan tertentu tidak ada operator untuk database favorit Anda, Anda dapat menggunakan JdbcOperator yang lebih umum atau menulis sendiri, Airflow mengizinkannya.
• 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, , , , .

Bagaimana Berhasil dalam Pengembangan Aliran Udara

• Berguna untuk menyimpan setiap pipeline (atau generator pipeline, lebih lanjut di bawah) dalam file terpisah. Saya segera tahu file mana yang harus saya tuju untuk melihat pipeline atau generator yang diperlukan.
• , , . , -, . , - , . : , , .
• – 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 titik di blok menunggu berubah menjadi dua sensor yang menunggu eksekusi pipa dasar,
• 4 tugas yang kami tentukan di blok buruh pelabuhan berubah menjadi 4 tugas yang berjalan secara paralel,
• kami menambahkan DummyOperator antara dua blok operator sehingga tidak ada jaringan koneksi antar tugas.

Apa yang ingin kami lakukan selanjutnya

Mengapa kami memilih Airflow

1. Pertama, ini adalah Python, di mana dengan bantuan dua loop dan beberapa kondisi, Anda dapat membuat pipeline yang elegan dan berfungsi dengan benar. Dan itu tidak perlu dijelaskan dalam XML yang sangat besar. Plus, hampir seluruh ekosistem Python dan seluruh perpustakaan kebun binatangnya tersedia di luar kotak, yang dapat digunakan sesuka Anda.
2. Tidak adanya XML sangat menyederhanakan tinjauan kode. Kami menulis kode pipeline dan konfigurasinya, dan semuanya bagus, semuanya berfungsi. Sebenarnya, Anda dapat menyeret XML atau format konfigurasi lainnya, tapi itu masalah selera.
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 ().