Основные концепции Airflow

Построение пайплайнов данных с Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Знакомство с инструктором

Фото профиля Volker Janz

Volker Janz

Senior Developer Advocate, Astronomer
14+ лет в роли дата-инженера в игровой индустрии
Работает с Airflow с версии 1.x
Спикер, ментор и ведущий рассылки Data Engineer Things

Что вы создадите

Создавайте DAG с помощью TaskFlow API
Строите динамические рабочие процессы с маппингом задач и расписанием на основе ресурсов
Обрабатывайте сбои с помощью повторных попыток и коллбэков
Запускайте SQL-нагрузки через Airflow

Визуализация содержания каждой главы

Перед началом

Понимание DAG, задач и операторов
Знакомство с основами расписания

Страница курса «Introduction to 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 — набор задач с зависимостями
Задачи — отдельные единицы работы
Операторы / декораторы определяют, что делает каждая задача
Зависимости задают порядок выполнения

Простой DAG

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

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

Оркестрация: Scheduler, Dag Processor
Выполнение: Worker, Triggerer
Интерфейс и хранилище: API Server, Metadata DB

Подходы к расписанию

# No automatic runs: trigger manually (default)
@dag(schedule=None)
def my_pipeline(): ...


# Time-based: runs every day at 6 AM
@dag(schedule="0 6 * * *")
def daily_pipeline(): ...


# Data-aware: runs when an Asset updates
@dag(schedule=[Asset("my_asset")])
def downstream_pipeline(): ...

Два способа написания DAG

Классические операторы

extract = PythonOperator(
    task_id="extract",
    python_callable=extract_fn)
extract >> transform

Используйте, когда декораторы недоступны

TaskFlow API

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

data = extract()
transform(data)

Простые декораторы Python
Меньше шаблонного кода
Совместимы с классическими операторами

Упражнения в этом курсе

Скриншот упражнения в IDE

Упражнения в IDE: редактируйте реальные файлы .py
Нажмите «Run this file» или используйте python3 filename.py
dag.test() запускает весь DAG в одном процессе

Давайте потренируемся!

Построение пайплайнов данных с Airflow