ddmitry/airflow-greenplum

Fork 0

Files

T

ddadmin 272013774b Сохранение логов airflow между перезапусками

2026-01-09 13:37:14 +03:00

32 KiB

Raw Permalink Blame History

airflow-dwh-gp-lab

Учебный стенд: ETL из Postgres в Greenplum с оркестрацией в Airflow.

Добро пожаловать в учебный стенд для изучения основ Data Engineering! Этот проект поможет вам освоить ключевые инструменты современных data pipeline: Airflow для оркестрации, pandas/CSV для подготовки данных, Postgres с демобазой bookings как источник и Greenplum как аналитическую базу данных.

Если вы проходите стенд как серию лабораторных, смотрите также файл с заданиями: educational-tasks.md.

🎯 Что вы узнаете

Как настроить локальный стек данных с помощью Docker
Как Airflow управляет workflow и координирует задачи
Как генерировать датасеты через pandas и сохранять их в CSV
Как загружать данные в Greenplum пакетами и избегать дублей
Как проверять качество данных в автоматизированных pipeline
Основы проектирования ETL/ELT процессов
Как работать с демо-БД bookings в Postgres как источником для будущего DWH

👩‍🎓 Для студентов (10‑минутный чек‑лист)

Установите Docker Desktop и Git.
Скопируйте настройки: cp .env.example .env.
Поднимите стенд: make up (или docker compose up -d — сервис airflow-init запустится автоматически при первом старте).
Откройте UI: http://localhost:8080 (admin/admin).
Включите и запустите DAG csv_to_greenplum. Дождитесь Success.
Проверьте данные: make gp-psql → SELECT COUNT(*) FROM public.orders;.
Дополнительно: запустите csv_to_greenplum_dq — все проверки должны быть зелёные.

Если что‑то не работает — смотрите «Типичные проблемы» и «Быстрый reset» ниже.

🚀 Быстрый старт (для новичков)

Шаг 1: Подготовка окружения

Требования:

Docker Desktop (Windows/Mac) или Docker Engine 24+ (Linux)
Git для клонирования репозитория

💡 Совет: Если у вас Windows, рекомендуем использовать WSL (Windows Subsystem for Linux) для лучшей совместимости.

Шаг 2: Настройка проекта

# Скопируйте файл настроек
cp .env.example .env

# Запустите стек (это может занять 2-3 минуты при первом запуске)
docker compose up -d

Шаг 3: Первый запуск pipeline

Откройте Airflow UI: http://localhost:8080 (логин/пароль: admin/admin)
Найдите DAG с названием csv_to_greenplum
Нажмите на переключатель слева от названия DAG, чтобы включить его
Нажмите кнопку Trigger (значок воспроизведения ▶️)

🎉 Поздравляем! Вы только что запустили свой первый data pipeline:

Система сгенерировала 1000 тестовых заказов при помощи pandas
Датасет сохранился в CSV-файл в каталоге ./data
Airflow загрузил данные из CSV в Greenplum без дублей по order_id

Шаг 4: Проверка результатов

Проверка вручную:

# Подключитесь к Greenplum и проверьте данные
docker compose exec greenplum bash -c "su - gpadmin -c 'psql -p 5432 -d gp_dwh'"

# Внутри psql выполните:
\dt                    # Показать таблицы
SELECT count(*) FROM public.orders;  # Посчитать записи

# Посмотреть несколько строк
SELECT * FROM public.orders LIMIT 5;

CSV-файлы после выполнения DAG остаются в директории ./data. Их можно открыть любым редактором или изучить через pandas.

Быстрый reset

Если после изменений что‑то «сломалось»:

make down                 # Остановить и удалить контейнеры/сети (volumes сохраняются)
make up

Если проблема связана с «грязной» остановкой и данными в томах (например, Greenplum не стартует), используйте полный reset: make clean && make up (данные в Docker-томах будут потеряны).

🛠️ Подробная настройка (для уверенных пользователей)

Если вы впервые запускаете стенд, этот раздел можно пролистать и вернуться к нему позже.

Установка Make (опционально)

Для удобства работы с проектом рекомендуем установить make:

Linux (Debian/Ubuntu): sudo apt install -y make
macOS: brew install make
Windows:
- WSL: sudo apt install -y make
- Chocolatey: choco install make
- Scoop: scoop install make

С make команды становятся короче:

make up                         # Запуск стека (включая airflow-init при первом старте)
make logs                       # Просмотр логов
make gp-psql                    # Подключение к Greenplum

Настройка подключения к Greenplum в Airflow

По умолчанию готовые DAG используют Airflow Connections. В docker-compose они заводятся автоматически через переменные окружения:

AIRFLOW_CONN_GREENPLUM_CONN — подключение к Greenplum с conn_id=greenplum_conn;
AIRFLOW_CONN_BOOKINGS_DB — подключение к демо-БД bookings с conn_id=bookings_db.

Такие подключения подхватываются из окружения и могут не отображаться в UI, но для DAG это нормально — PostgresOperator найдёт их по conn_id.

При желании вы можете создать или отредактировать подключение вручную в UI:

Airflow UI → Admin → Connections → Add a new record
Заполните поля:
- Conn Id: greenplum_conn
- Conn Type: Postgres
- Host: greenplum
- Schema: gp_dwh
- Login: gpadmin
- Password: gpadmin
- Port: 5432

Greenplum + PXF: свой образ

Чтобы избежать проблем с правами и нестабильных запусков, Greenplum собирается из собственного Dockerfile.greenplum. В образ вшиты:

JDBC-драйвер PostgreSQL;
конфигурация PXF-сервера bookings-db;
ensure-скрипт, который при каждом старте контейнера докладывает файлы в PXF_BASE.

Дополнительно при старте контейнера:

базовые конфиги PXF копируются в PXF_BASE/conf (если их ещё нет);
создаются каталоги PXF_BASE/run и PXF_BASE/logs;
CREATE EXTENSION pxf выполняется автоматически, когда Greenplum становится доступен (с ретраями).

Healthcheck сервиса greenplum учитывает не только готовность Greenplum, но и запуск PXF, а также наличие extension pxf — это нужно, чтобы Airflow не стартовал раньше PXF.

Сборка и запуск:

make build — собрать образ (явно);
make up — поднимет стек и соберёт образ, если он ещё не создан.

Обновление PXF-конфигов:

изменили файлы в pxf/ → выполните make build и перезапустите контейнер;
для принудительной перезаписи файлов в PXF_BASE используйте PXF_SEED_OVERWRITE=1;
для принудительного pxf cluster sync при старте используйте PXF_SYNC_ON_START=1.

Проверка PXF:

статус: docker compose exec greenplum bash -lc "su - gpadmin -c '/usr/local/pxf/bin/pxf cluster status'";
логи: greenplum_data:/data/pxf/logs (внутри контейнера — /data/pxf/logs).

Локальное окружение разработчика

Локальным окружением управляет uv — он скачивает нужный Python и создаёт .venv на основе pyproject.toml / uv.lock.

uv sync

uv sync сам подтянет версию Python из .python-version/pyproject.toml, создаст .venv и установит зависимости. Для тех же действий можно использовать make dev-sync. Цель make dev-setup (или вручную uv python install + uv python pin) нужна только когда вы меняете версию Python или прогреваете кэш.

Если требуется «классическое» активированное окружение, после uv sync выполните .\.venv\Scripts\Activate.ps1 в PowerShell или source .venv/bin/activate в Unix-терминале.

Проверки и форматирование выполняем через uv:

make test            # uv run pytest -q
make lint            # black/isort в режиме проверки
make fmt             # автоформатирование black + isort

Быстрый старт с uv

uv sync
uv run pytest -q
uv run black --check airflow tests

Не устанавливайте пакеты напрямую через pip install --user .... Если что-то уже попало в user-site, удалите pip uninstall <package> и проверьте pip list --user.

📋 Что входит в стенд

Основные компоненты

Greenplum — аналитическая база данных для хранения и анализа данных
Airflow — оркестратор workflow и задач
Postgres — база метаданных для Airflow
Postgres (bookings) — отдельная демо-БД bookings (источник данных для будущего DWH в Greenplum)
pandas — библиотека для генерации и анализа данных в формате CSV

Готовые DAG (workflow)

orders_base_ddl — создаёт базовую таблицу public.orders для CSV‑пайплайна
bookings_stg_ddl — готовит схему stg и таблицы stg.bookings_ext / stg.bookings
csv_to_greenplum — базовый pipeline: pandas → CSV → Greenplum
bookings_to_gp_stage — пример загрузки из демо‑БД bookings в слой STG
csv_to_greenplum_dq — проверки качества данных (наличие таблицы, схема, дубликаты)

Учебный путь — триггернуть DDL‑DAG: для CSV orders_base_ddl, для bookings bookings_stg_ddl. Технический шорткат для быстрой инициализации — make ddl-gp (он не вызывается автоматически при старте контейнеров).

Полезные команды

# Основные команды
make up                 # Запустить весь стенд (Airflow инициализируется автоматически при первом старте)
make stop               # Остановить контейнеры, не трогая данные
make down               # Остановить и удалить контейнеры/сети (volumes сохраняются)
make clean              # Полный reset: остановить и удалить контейнеры/сети и тома (данные будут потеряны)
make airflow-init       # Ручной запуск инициализации Airflow (обычно не нужен)
make ddl-gp             # Применить DDL к Greenplum вручную
make gp-psql            # Подключиться к Greenplum через psql
make bookings-init      # Установить демобазу bookings в Postgres (по умолчанию генерирует 1 день)
make bookings-generate-day  # Добавить ещё один день данных в bookings (можно вызвать несколько раз)
make bookings-psql      # Подключиться к демобазе bookings (БД demo)

# Проверка данных
make logs               # Следить за логами Airflow

# Логи задач Airflow сохраняются в Docker-томе `airflow_logs`
# и переживают `docker compose down`/`up` (удаляются при `docker compose down -v` / `make clean`).

# Контроль генерации bookings
docker compose -f docker-compose.yml exec bookings-db bash -lc 'PGPASSWORD="$POSTGRES_PASSWORD" psql -U "$POSTGRES_USER" -d demo -c "SELECT busy();"'
# busy() = t — генерация ещё идёт; f — завершена. При необходимости можно вызвать CALL abort(); и запустить генерацию заново.

Генерация следующего дня в bookings

При make bookings-init автоматически генерируется BOOKINGS_INIT_DAYS суток, начиная с даты BOOKINGS_START_DATE (по умолчанию один день с 2017-01-01).
Дальше каждый вызов make bookings-generate-day или генерации через DAG добавляет ровно один следующий день после max(book_date) в bookings.bookings — генератор сам смотрит последнюю дату.
Рекомендуемый учебный сценарий для DAG bookings_to_gp_stage: запускать DAG по одному дню вперёд, выбирая в форме Trigger логическую дату Execution Date (ds), совпадающую с тем днём, который вы хотите загрузить (например, 2017-01-01, затем 2017-01-02 и т.д.).
Быстрее всего: make bookings-generate-day — читает GUC и сам вызывает continue.

Вручную из psql/DBeaver (подзапросы в аргументах CALL не работают, поэтому через DO-блок):

DO $$
DECLARE
  v_next_day timestamptz;
BEGIN
  SELECT date_trunc('day', max(book_date)) + interval '1 day'
  INTO v_next_day
  FROM bookings.bookings;
  CALL continue(v_next_day);           -- или CALL continue(v_next_day, 4) для параллельности
END $$;

Не вызывайте CALL generate(...) поверх существующих данных: она делает TRUNCATE и создаёт демобазу заново.

⚙️ Настройка через переменные окружения

Все настройки находятся в файле .env. Основные параметры:

Greenplum

GP_USER — пользователь (по умолчанию: gpadmin)
GP_PASSWORD — пароль (по умолчанию: gpadmin)
GP_DB — база данных (по умолчанию: gp_dwh)
GP_PORT — порт Greenplum внутри Docker-сети (по умолчанию: 5432, менять обычно не нужно; внешний порт на хосте для подключения клиентов — 5435).

Демо-БД bookings (Postgres)

BOOKINGS_DB_USER — пользователь Postgres для демобазы (по умолчанию: bookings)
BOOKINGS_DB_PASSWORD — пароль пользователя (по умолчанию: bookings)
BOOKINGS_DB_NAME — база данных, из которой запускается установка генератора (по умолчанию: bookings)
BOOKINGS_DB_PORT — внешний порт для подключения к контейнеру bookings-db (по умолчанию: 5434)
BOOKINGS_START_DATE — начальная дата модельного времени (по умолчанию: 2017-01-01)
BOOKINGS_INIT_DAYS — сколько дней сгенерировать при первой инициализации (по умолчанию: 1, чтобы увидеть данные без долгого ожидания; можно менять при вызове BOOKINGS_INIT_DAYS=... make bookings-init)
BOOKINGS_JOBS — число параллельных джобов генератора bookings (по умолчанию: 1; при 1 генерация идёт синхронно без dblink)

CSV pipeline

CSV_DIR — путь к каталогу с CSV внутри контейнеров Airflow (по умолчанию: /opt/airflow/data)
CSV_ROWS — количество строк, генерируемых DAG (по умолчанию: 1000)

Airflow

GP_CONN_ID — ID подключения (по умолчанию: greenplum_conn)

🔍 Продвинутые темы

Этот раздел не обязателен при первом прохождении стенда; к нему удобно вернуться, когда базовый CSV‑pipeline уже понятен.

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

Поток данных в DAG csv_to_greenplum:

create_orders_table — создаёт таблицу public.orders в Greenplum
generate_csv — генерирует датасет при помощи pandas и сохраняет CSV в CSV_DIR
preview_csv — выводит предпросмотр и статистику по данным
load_csv_to_greenplum — загружает CSV во временную таблицу и переносит новые строки в public.orders

💡 Безопасность повторного запуска: Pipeline защищен от дубликатов, поэтому его можно запускать многократно.

Проверка качества данных

Запустите DAG csv_to_greenplum_dq для автоматической проверки:

Наличие таблицы в базе
Соответствие схемы ожидаемой структуре
Объем загруженных данных
Отсутствие дубликатов записей

Пример DAG с SQL-скриптами (bookings → stg)

Если вы ещё не дошли до части про bookings и слои DWH, этот подраздел можно пропустить на первом чтении.

В репозитории есть учебный DAG bookings_to_gp_stage, который показывает «канонический» способ работы с SQL в Airflow:

подключение к БД через Airflow Connections (bookings_db, greenplum_conn);
бизнес-логика инкрементальной загрузки и DQ вынесена в SQL-файлы в каталоге sql/:
- sql/src/bookings_generate_day_if_missing.sql — генерация следующего учебного дня в демо-БД bookings (или нескольких стартовых дней, если база пуста);
- sql/stg/bookings_ddl.sql — DDL для схемы stg и таблиц stg.bookings_ext / stg.bookings;
- sql/stg/bookings_load.sql — загрузка инкремента из stg.bookings_ext в stg.bookings на основе «хвоста» после предыдущих батчей;
- sql/stg/bookings_dq.sql — проверка количества строк между источником и stg за то же окно.

Фрагмент DAG:

load_bookings_to_stg = PostgresOperator(
    task_id="load_bookings_to_stg",
    postgres_conn_id="greenplum_conn",
    sql="stg/bookings_load.sql",
    params={"batch_id": "{{ ds_nodash }}"},
)

Такой подход помогает держать оркестрацию (DAG) и SQL-логику в отдельных файлах и легче сравнивать её с теорией из статьи про моделирование DWH.

Ограничения учебного стенда

Greenplum запущен в single-node режиме (для обучения)
В продакшене Greenplum обычно разворачивают кластером на нескольких серверах
Используется Greenplum 6 (широко доступная версия), хотя Greenplum 7 предлагает больше возможностей

🧩 Подключение к базам через DBeaver

Необязательный раздел: нужен только если вы хотите смотреть данные через DBeaver. Для базовых заданий достаточно make gp-psql.

Ниже — краткая инструкция, как подключиться к Greenplum и демо-БД bookings из DBeaver. Перед этим убедитесь, что стенд запущен:

cp .env.example .env (если ещё не делали)
make up
make bookings-init
make ddl-gp

Greenplum (аналитическая БД)

Откройте DBeaver → New Database Connection.
Выберите драйвер PostgreSQL (или Greenplum, если он есть в вашей версии DBeaver).
На вкладке Main заполните поля (по умолчанию):
- Host: localhost
- Port: 5435 (внешний порт Greenplum на хосте)
- Database: значение GP_DB (по умолчанию gp_dwh)
- Username: значение GP_USER (по умолчанию gpadmin)
- Password: значение GP_PASSWORD (по умолчанию gpadmin)
Нажмите Test Connection → OK, затем Finish.

После подключения:

Основные таблицы лаба — в схеме public базы gp_dwh (например, public.orders).
После настройки PXF и выполнения make ddl-gp станет доступна внешняя таблица public.ext_bookings_bookings — чтение из демо-БД bookings через PXF.

bookings-db (демо-БД источника)

Для работы с исходными данными (демо-БД demo) достаточно стандартного PostgreSQL-подключения.

Откройте DBeaver → New Database Connection → драйвер PostgreSQL.
На вкладке Main укажите (значения по умолчанию из .env.example):
- Host: localhost
- Port: значение BOOKINGS_DB_PORT из .env (по умолчанию 5434)
- Database: demo
- Username: BOOKINGS_DB_USER (по умолчанию bookings)
- Password: BOOKINGS_DB_PASSWORD (по умолчанию bookings)
Нажмите Test Connection → OK, затем Finish.