Базові поняття Airflow

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Познайомтесь з інструктором

  Фото профілю Volker Janz

$$

Volker Janz

$$

  • Senior Developer Advocate, Astronomer
  • 14+ років дата-інженерії в ґеймінгу
  • Працює з Airflow з версії 1.x
  • Спікер, ментор і редактор розсилки в Data Engineer Things
Створення конвеєрів даних з Airflow

Що ви створите

 

  • Створюйте Dags за допомогою TaskFlow API
  • Будуйте динамічні робочі процеси через мапінг завдань і планування на основі активів
  • Обробляйте збої за допомогою повторів і зворотних викликів
  • Запускайте навантаження SQL через Airflow

Візуалізація змісту кожного розділу

Створення конвеєрів даних з Airflow

Перш ніж почати

 

$$

  • Впевнено працюватимете з Dags, tasks і операторами
  • Розумітимете основи планування

Introduction to Airflow - сторінка курсу

Створення конвеєрів даних з Airflow

Коротке нагадування

from airflow.sdk import dag, task

@dag
def star_wars_dag():


  @task
  def get_star_wars_person():
    import requests
    return requests.get("https://swapi.dev/api/people/1/").json()


  @task.bash
  def print_name(person):
    return f"echo '{person['name']}'"


  person = get_star_wars_person()
  print_name(person)

star_wars_dag()
  • Dag — це набір завдань із залежностями
  • Завдання (tasks) — окремі одиниці роботи
  • Оператори / декоратори визначають, що робить кожне завдання
  • Залежності визначають порядок виконання

$$

Простий Dag

Створення конвеєрів даних з Airflow

Архітектура Airflow

$$

Архітектура Airflow 3

 

$$

  • Оркестрація: Scheduler, Dag Processor
  • Виконання: Worker, Triggerer
  • Інтерфейс і сховище: API Server, Metadata DB
Створення конвеєрів даних з Airflow

Підходи до планування

# Автоматичні запуски вимкнено: запуск вручну (типово)
@dag(schedule=None)
def my_pipeline(): ...


# За часом: щодня о 6:00
@dag(schedule="0 6 * * *")
def daily_pipeline(): ...


# З урахуванням даних: запускається, коли оновлюється Asset
@dag(schedule=[Asset("my_asset")])
def downstream_pipeline(): ...
Створення конвеєрів даних з Airflow

Два способи писати Dags

Класичні оператори

extract = PythonOperator(
    task_id="extract",
    python_callable=extract_fn)
extract >> transform
  • Обирайте, коли немає наявних декораторів

TaskFlow API

@task
def extract():
    return {"users": 150}

data = extract()
transform(data)
  • Прості декоратори Python
  • Менше шаблонного коду
  • Можна поєднувати з класичними операторами
Створення конвеєрів даних з Airflow

Вправи в цьому курсі

$$

Скріншот вправи в IDE

 

  • Вправи в IDE: редагуєте справжні файли .py
  • Натисніть "Run this file" або використайте python3 filename.py
  • dag.test() запускає весь Dag в одному процесі
Створення конвеєрів даних з Airflow

Давайте потренуємось!

Створення конвеєрів даних з Airflow

Preparing Video For Download...