ETL-пайплайн: CSV в базу данных

Пора собрать всё вместе. Приложение «Гонца» каждую ночь выгружает заказы в CSV-файл: время заказа по Екатеринбургу, время доставки в UTC, отменённые вперемешку с доставленными. Утром у Отто должны быть таблица доставок в хранилище и витрина по зонам. Пайплайн должен переживать моргнувшую сеть, не плодить дубли при перезапуске, останавливаться на кривых данных и уметь перезалить любую прошлую неделю. Каждую из этих деталей мы уже делали по отдельности. В этом уроке они складываются в один файл.

Конвейер

Хлебозавод работает по конвейеру. Мука приходит мешками, на первом посту её просеивают, на втором замешивают тесто, на третьем лаборант проверяет его на влажность, на четвёртом пекут. Каждый пост делает одно дело и передаёт результат дальше. Если тесто не прошло проверку, его не отправляют в печь. А если ночью печь остановилась на полчаса, утром смену не начинают с нуля, а допекают то, что не допекли.

Ночной пайплайн устроен так же: каждая задача делает одно дело, передаёт результат следующей, а проверка стоит перед загрузкой.

Весь файл

import csv
import sqlite3
from datetime import datetime, timedelta, timezone

from airflow.exceptions import AirflowFailException
from airflow.sdk import DAG, task
from airflow.timetables.interval import CronDataIntervalTimetable

DB = "warehouse.db"
YEKATERINBURG = timezone(timedelta(hours=5))


def created_utc(row):
    return datetime.fromisoformat(row["created_at"]).replace(tzinfo=YEKATERINBURG).astimezone(timezone.utc)


def delivered_utc(row):
    if not row["delivered_at"]:
        return None
    return datetime.fromisoformat(row["delivered_at"].removesuffix("+00")).replace(tzinfo=timezone.utc)


with DAG(
    dag_id="gonets_daily",
    schedule=CronDataIntervalTimetable("@daily", timezone="UTC"),   # 1
    start_date=datetime(2026, 10, 1, tzinfo=timezone.utc),
    catchup=True,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},   # 2
) as dag:

    @task
    def extract_orders(data_interval_start=None, data_interval_end=None):   # 3
        with open("orders.csv", encoding="utf-8") as file:
            return [row for row in csv.DictReader(file)
                    if data_interval_start <= created_utc(row) < data_interval_end]

    @task
    def clean(rows):   # 4
        ...

    @task
    def check(rows):   # 5
        if not rows:
            raise AirflowFailException("после чистки не осталось ни одной доставки")
        ...
        return rows

    @task
    def load(rows, data_interval_start=None):   # 6
        day = data_interval_start.date().isoformat()
        conn = sqlite3.connect(DB)
        with conn:
            conn.execute("DELETE FROM fact_deliveries WHERE day = ?", (day,))
            conn.executemany("INSERT INTO fact_deliveries VALUES (...)", rows)
        conn.close()
        return len(rows)

    load(check(clean(extract_orders())))   # 7

Пройдём по номерам:

1. Суточный интервал. Запуск за 3 октября отвечает за заказы с 3 октября 00:00 до 4 октября 00:00 по UTC и стартует после конца своих суток. Почему не "@daily", разбирали в уроке про backfill.
2. Повторы. Моргнувшая сеть не роняет ночь, задача повторится через пять минут.
3. Границы из запуска, а не из часов. Поэтому backfill за прошлую неделю обработает именно ту неделю.
4. Чистка. Переводит время заказа в UTC, отбрасывает отменённые заказы и доставки без времени, считает минуты доставки. Это шаг T из ETL.
5. Проверка. Бросает AirflowFailException, если данные кривые: повторять бессмысленно.
6. Идемпотентная загрузка. Удалить свой день и вставить заново в одной транзакции, как в уроке 15.
7. Цепочка. Каждая задача получает результат предыдущей через XCom, и порядок задаётся этим же вызовом.

Через XCom здесь едут сами строки заказов. Для трёхсот строк за день это допустимо. Если бы заказов были десятки тысяч, чистка писала бы результат во временный файл или таблицу, а дальше передавала только путь.