318 lines
18 KiB
Markdown
318 lines
18 KiB
Markdown
# Восстановите прием справочника из источника
|
||
|
||
Хранилище получает справочник аэропортов из Postgres. Названия и другие
|
||
атрибуты в источнике могут меняться, поэтому загрузка должна сохранять
|
||
состояние справочника на момент приема данных.
|
||
|
||
Вы разберете готовый участок `airports` и напишете его заново: создадите
|
||
внешнюю таблицу PXF и загрузку полного снимка в STG. Затем испытаете ее
|
||
на повторном запуске и изменении источника. Полученный снимок понадобится
|
||
следующему слою — ODS.
|
||
|
||
## Задача
|
||
|
||
Ваш участок на [карте](../design/architecture-map.html#node=stg.airports):
|
||
`bookings.airports_data → stg.airports_ext → stg.airports`.
|
||
|
||
| Что написать | Где |
|
||
|---|---|
|
||
| DDL внешней таблицы `stg.airports_ext` | [airports_ddl.sql](../../sql/stg/airports_ddl.sql) |
|
||
| Загрузку полного снимка во внутреннюю `stg.airports` | [airports_load.sql](../../sql/stg/airports_load.sql) |
|
||
|
||
Внутренняя таблица, JDBC-сервер PXF, DAG и проверки качества уже готовы.
|
||
Рядом есть пример для другого справочника:
|
||
[DDL airplanes](../../sql/stg/airplanes_ddl.sql) и
|
||
[его загрузка](../../sql/stg/airplanes_load.sql). Используйте его, учитывая
|
||
состав полей и ключ аэропорта.
|
||
|
||
Требования к вашей загрузке:
|
||
|
||
- Источник — `bookings.airports_data` в базе `demo`, ключ — `airport_code`.
|
||
- Все шесть полей (`airport_code`, `airport_name`, `city`, `country`,
|
||
`coordinates`, `timezone`) попадают в STG без потерь и преобразований.
|
||
Внешняя и внутренняя таблицы используют TEXT; JSON и координаты
|
||
пока хранятся как текст.
|
||
- Строки одного снимка получают `_load_id = '{{ run_id }}'`,
|
||
`_load_ts = now()` и `event_ts = now()`. Точного времени изменения
|
||
аэропорта в источнике нет, поэтому `event_ts` — время получения снимка.
|
||
- Новый запуск сохраняет новый полный снимок, оставляя предыдущие.
|
||
Повторное выполнение загрузки в том же запуске на неизменном источнике
|
||
не меняет бизнес-данные и не создает дубли ключей внутри снимка.
|
||
|
||
## Подготовка
|
||
|
||
Нужен свой стенд после [быстрого старта и знакомства с моделью](README.md#подготовка).
|
||
Во время работы не запускайте другие загрузки: они могут изменить данные
|
||
посреди опыта. Если запущен `bookings_validate`, дождитесь его завершения.
|
||
|
||
Сначала освободите участок для своей реализации. Команды ниже удаляют
|
||
внешнюю таблицу и **всю историю только `stg.airports`**. Это нужно, чтобы
|
||
данные быстрого старта не скрыли отсутствие вашей загрузки.
|
||
|
||
<details>
|
||
<summary>Подготовка файлов и таблиц</summary>
|
||
|
||
Сохраните свои SQL-файлы в Git или отдельной копии. В `airports_ddl.sql`
|
||
удалите блок от `DROP EXTERNAL TABLE` до конца `CREATE EXTERNAL TABLE`;
|
||
DDL схемы и внутренней таблицы оставьте. Содержимое `airports_load.sql`
|
||
замените на `SELECT 1;`. DAG и DQ менять не нужно.
|
||
|
||
В Greenplum (`make gp-psql`):
|
||
|
||
```sql
|
||
DROP EXTERNAL TABLE IF EXISTS stg.airports_ext;
|
||
TRUNCATE TABLE stg.airports;
|
||
SELECT COUNT(*) FROM stg.airports; -- 0
|
||
```
|
||
|
||
Остальные справочники и текущий ODS сохраняются. Пока участок
|
||
не восстановлен, ODS запускать не нужно.
|
||
|
||
</details>
|
||
|
||
## Соберите загрузку
|
||
|
||
Начните с внешней таблицы. Добейтесь, чтобы запрос к `stg.airports_ext`
|
||
возвращал те же поля и значения, что запрос к `bookings.airports_data`
|
||
в Postgres. Для начала достаточно рассмотреть несколько аэропортов.
|
||
|
||
<details>
|
||
<summary>Ориентиры для PXF и применение DDL</summary>
|
||
|
||
В `LOCATION` укажите таблицу источника, `PROFILE=JDBC` и `SERVER=bookings-db`.
|
||
`SERVER` — имя готового [каталога конфигурации PXF](../../pxf/servers/bookings-db/jdbc-site.xml).
|
||
Формат чтения — `FORMAT 'CUSTOM' (formatter='pxfwritable_import')`.
|
||
Логин и пароль в DDL не нужны. Если меняли учетные данные источника,
|
||
согласуйте их с PXF по [справочнику стенда](../stack.md).
|
||
|
||
Примените свой файл из `make gp-psql`:
|
||
|
||
```sql
|
||
\set ON_ERROR_STOP on
|
||
\i /sql/stg/airports_ddl.sql
|
||
SELECT * FROM stg.airports_ext ORDER BY airport_code LIMIT 3;
|
||
```
|
||
|
||
В Postgres (`make bookings-psql`) выполните такой же запрос к
|
||
`bookings.airports_data`. Сверьте состав и порядок полей с DDL.
|
||
Если чтение не работает, начните с сообщения об ошибке и
|
||
[диагностики STG](../bookings_to_gp_stage.md#если-загрузка-не-прошла).
|
||
|
||
</details>
|
||
|
||
Теперь напишите загрузку снимка. Продумайте, как ваш SQL отличит
|
||
повторное выполнение от нового запуска и сохранит историю.
|
||
Выполнять загрузку будет готовый DAG `lab_pxf_airports`.
|
||
|
||
Получите первый снимок и сравните его с источником готовой сверкой.
|
||
Она проверяет ключи, все бизнес-поля и служебные метки. Зеленые задачи
|
||
в Airflow показывают, что код выполнился; сверка помогает обнаружить
|
||
ошибку в самих данных.
|
||
|
||
<details>
|
||
<summary>Первый запуск в Airflow и сверка с Postgres</summary>
|
||
|
||
Для сверки нужен `uv` с Python 3.11 из
|
||
[окружения разработчика](../stack.md#локальное-окружение-разработчика-uv).
|
||
В терминале из корня репозитория сохраните исходный справочник:
|
||
|
||
```bash
|
||
mkdir -p data/lab9
|
||
uv run --no-project scripts/check_airports_snapshot.py export data/lab9/airports-before.json
|
||
```
|
||
|
||
Скрипт читает Postgres напрямую, минуя вашу внешнюю таблицу, и не перезаписывает
|
||
существующий файл. Эта копия позволит сравнить первый снимок с источником
|
||
даже после изменения названия аэропорта.
|
||
|
||
Откройте Airflow UI по адресу из `.env` (`AIRFLOW_WEB_PORT`), войдите с
|
||
`AIRFLOW_USER` и `AIRFLOW_PASSWORD`. Включите `lab_pxf_airports`, нажмите
|
||
**Trigger DAG** и дождитесь успеха всех задач. В **Grid** выберите запуск;
|
||
его **Run ID** подставьте в команду:
|
||
|
||
```bash
|
||
first_run='вставьте Run ID первого запуска'
|
||
uv run --no-project scripts/check_airports_snapshot.py check "$first_run" data/lab9/airports-before.json
|
||
```
|
||
|
||
Если задача упала, откройте ее **Log**. SQL-файлы доступны контейнеру
|
||
из рабочей папки: после правки пересобирать образ не нужно.
|
||
`{{ run_id }}` подставляет Airflow, поэтому файл загрузки не следует
|
||
выполнять напрямую через `psql`.
|
||
|
||
</details>
|
||
|
||
## Испытайте загрузку
|
||
|
||
### Повторное выполнение
|
||
|
||
Представьте, что после сбоя пришлось повторить уже выполненную задачу.
|
||
Что произойдет с данными при повторном выполнении вашего SQL? За счет
|
||
чего в одном снимке не появится вторая копия каждого аэропорта?
|
||
|
||
Повторите загрузку **в том же запуске** и снова выполните сверку первого
|
||
снимка. При ошибке исправьте SQL и повторите опыт на том же Run ID.
|
||
|
||
<details>
|
||
<summary>Как повторить задачу, сохранив Run ID</summary>
|
||
|
||
В запуске первого снимка выберите `load_airports_to_stg` → **Clear task**.
|
||
Снимите **Upstream**, **Downstream**, **Past**, **Future**, **Recursive**
|
||
и **Only Failed**, если включены. В **Affected Tasks** должна остаться
|
||
только эта задача этого запуска. Нажмите **Clear**, дождитесь успеха,
|
||
затем так же повторите `check_airports_dq`.
|
||
|
||
Новый **Trigger DAG** создал бы другой Run ID. Здесь нужно повторное
|
||
выполнение с прежним `_load_id`, чтобы испытать защиту от дублей.
|
||
|
||
Если загрузка уже записала дубли или неверные значения, исправление SQL
|
||
само их не удалит.
|
||
Удалите из `stg.airports` строки только этого `_load_id`, затем повторите
|
||
исправленную загрузку и DQ в том же запуске. После успешной сверки повторите
|
||
их еще раз: теперь вы проверяете защиту от дублей на заполненном снимке.
|
||
|
||
</details>
|
||
|
||
### Изменение в источнике
|
||
|
||
Теперь аэропорт переименовали. До новой загрузки предположите, где будет
|
||
видно новое название: в Postgres, во внешней таблице, в первом снимке STG?
|
||
|
||
Измените русское название одного аэропорта, сохранив ключ и остальные
|
||
языки. Для опыта ниже есть готовый SQL, который также сохраняет исходное
|
||
название для восстановления.
|
||
|
||
<details>
|
||
<summary>Изменить название в Postgres</summary>
|
||
|
||
В `make bookings-psql`:
|
||
|
||
```sql
|
||
\set ON_ERROR_STOP on
|
||
CREATE TABLE public.lab9_airport_original AS
|
||
SELECT airport_code, airport_name
|
||
FROM bookings.airports_data ORDER BY airport_code LIMIT 1;
|
||
|
||
UPDATE bookings.airports_data AS a
|
||
SET airport_name = jsonb_set(
|
||
b.airport_name, '{ru}', to_jsonb((b.airport_name->>'ru') || ' [lab9]'), false
|
||
)
|
||
FROM public.lab9_airport_original AS b
|
||
WHERE a.airport_code = b.airport_code
|
||
RETURNING a.airport_code, b.airport_name AS original, a.airport_name AS changed;
|
||
```
|
||
|
||
Запрос изменяет один аэропорт. Если таблица копии уже существует после
|
||
предыдущего опыта, сначала [восстановите источник](#завершение-работы),
|
||
сверьте результат и удалите таблицу копии. Затем можно начать новый опыт.
|
||
|
||
</details>
|
||
|
||
Найдите этот аэропорт в `stg.airports_ext` и `stg.airports` обычными
|
||
SELECT-запросами. Сопоставьте наблюдение со своим предположением.
|
||
|
||
Запустите `lab_pxf_airports` заново и рассмотрите оба снимка в STG.
|
||
Какое название хранится в каждом? Почему одного сравнения количества
|
||
строк недостаточно, чтобы заметить переименование?
|
||
|
||
Сверьте новый снимок с измененным источником, а первый — с исходным.
|
||
Это обнаружит, например, загрузку старых данных под новым Run ID.
|
||
|
||
<details>
|
||
<summary>Сверить оба снимка</summary>
|
||
|
||
В том же терминале, где задан `first_run`:
|
||
|
||
```bash
|
||
uv run --no-project scripts/check_airports_snapshot.py export data/lab9/airports-after.json
|
||
next_run='вставьте Run ID нового запуска'
|
||
uv run --no-project scripts/check_airports_snapshot.py check "$next_run" data/lab9/airports-after.json
|
||
uv run --no-project scripts/check_airports_snapshot.py check "$first_run" data/lab9/airports-before.json
|
||
```
|
||
|
||
Если терминал закрывался, значения Run ID можно снова взять из Airflow.
|
||
|
||
</details>
|
||
|
||
## Передайте снимок в ODS
|
||
|
||
Запустите `bookings_to_gp_ods` без параметров `conf`. Найдите измененный
|
||
аэропорт в `ods.airports`: название должно быть новым, а `_load_id` —
|
||
совпадать с последним запуском STG. Так вы увидите, что следующий слой
|
||
использовал результат вашей загрузки.
|
||
|
||
ODS выбирает общий батч четырех справочников: `airports`, `airplanes`,
|
||
`routes`, `seats`. Поэтому готовый `lab_pxf_airports` загружает все четыре,
|
||
используя те же SQL, что основной STG-DAG. Генератор новых дней и
|
||
транзакционные таблицы в этом опыте не участвуют.
|
||
|
||
<details>
|
||
<summary>Проверить выбор снимка и данные ODS</summary>
|
||
|
||
В Log задачи `resolve_stg_batch_id` строка `Используем stg_batch_id=...`
|
||
показывает выбранный Run ID. Он должен совпасть с `next_run`.
|
||
После завершения DAG выполните:
|
||
|
||
```bash
|
||
uv run --no-project scripts/check_airports_snapshot.py check "$next_run" data/lab9/airports-after.json --layer ods
|
||
```
|
||
|
||
Сверка учитывает преобразование JSON в русские названия и требует нужный
|
||
`_load_id` у всех строк ODS. Если выбран старый батч, посмотрите, завершились
|
||
ли все загрузки и DQ последнего запуска `lab_pxf_airports`.
|
||
|
||
</details>
|
||
|
||
## Завершение работы
|
||
|
||
Верните исходное название и получите еще один общий снимок через
|
||
`lab_pxf_airports`. Проведите его по цепочке
|
||
`bookings_to_gp_ods` → `bookings_to_gp_dds` → `bookings_to_gp_dm`.
|
||
В `ods.airports` и `dds.dim_airports` должно вернуться исходное название.
|
||
|
||
<details>
|
||
<summary>Восстановить источник и сверить результат</summary>
|
||
|
||
В `make bookings-psql`:
|
||
|
||
```sql
|
||
UPDATE bookings.airports_data AS a
|
||
SET airport_name = b.airport_name
|
||
FROM public.lab9_airport_original AS b
|
||
WHERE a.airport_code = b.airport_code
|
||
RETURNING a.airport_code, a.airport_name;
|
||
```
|
||
|
||
После нового запуска `lab_pxf_airports` и ODS сравните восстановленный
|
||
снимок с исходной копией:
|
||
|
||
```bash
|
||
restored_run='вставьте Run ID запуска STG после восстановления'
|
||
uv run --no-project scripts/check_airports_snapshot.py check "$restored_run" data/lab9/airports-before.json
|
||
uv run --no-project scripts/check_airports_snapshot.py check "$restored_run" data/lab9/airports-before.json --layer ods
|
||
```
|
||
|
||
Повтор UPDATE восстановления безопасен. После успешной сверки выполните
|
||
в Postgres `DROP TABLE public.lab9_airport_original;`.
|
||
Если прервались на изменении источника,
|
||
продолжите с этого восстановления и нового запуска цепочки.
|
||
|
||
Для повторения всей лабораторной возьмите новые имена контрольных файлов.
|
||
При полной пересборке сохраните SQL-файлы: `make clean` удаляет тома.
|
||
Затем снова пройдите быстрый старт со своим решением.
|
||
|
||
</details>
|
||
|
||
Снимки в STG остаются историей опыта: первоначальный, с измененным названием
|
||
и восстановленный. Последний общий снимок и текущий ODS снова содержат
|
||
исходные значения; транзакционные данные остались от быстрого старта.
|
||
|
||
Результат работы — ваши два SQL-файла и объяснение их поведения: откуда
|
||
берется каждый снимок, что защищает его от дублей при повторе и как
|
||
изменение источника попадает в следующую загрузку.
|
||
|
||
Дальше — [загрузки `ods.airplanes` и `ods.seats`](README.md#первая-загрузка-ods).
|
||
Они используют уже подготовленный снимок, поэтому заново запускать STG
|
||
для начала следующей работы не нужно. Там вы будете преобразовывать поля
|
||
полученных данных и заменять текущее состояние в ODS.
|