ddadmin
2228312b55
- Зачем: - предоставить студентам полный набор аналитических витрин с примерами различных паттернов (UPSERT, Full Rebuild, UNION ALL, двухуровневая агрегация). - Что: - реализованы витрины: sales_report, route_performance, passenger_loyalty, airport_traffic, monthly_overview. - исправлен баг в route_performance_load.sql: добавлены JOIN к dim_airports и dim_airplanes для корректной денормализации атрибутов. - обновлен скрипт e2e_etl.sh: добавлена верификация всех 5 витрин и проверка бизнес-логики (load factor). - обновлены DAGи, тесты и главный DDL скрипт. - Проверка: - автоматизированный прогон e2e_etl.sh через REST API Airflow.
116 lines
4.5 KiB
Bash
Executable File
116 lines
4.5 KiB
Bash
Executable File
#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
|
|
# Скрипт автоматизированного прогона E2E-теста всей цепочки DWH через REST API Airflow.
|
|
# Отрабатывает 2 "учебных дня" для проверки инкрементальной загрузки.
|
|
|
|
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
|
cd "$ROOT"
|
|
|
|
warn() { echo ":: ${1}" >&2; }
|
|
|
|
# Загружаем переменные из .env, если файл существует, иначе используем дефолты
|
|
if [ -f ".env" ]; then
|
|
export $(grep -v '^#' .env | xargs)
|
|
fi
|
|
|
|
AIRFLOW_USER=${AIRFLOW_USER:-admin}
|
|
AIRFLOW_PASSWORD=${AIRFLOW_PASSWORD:-admin}
|
|
API_URL="http://localhost:8080/api/v1/dags"
|
|
|
|
# Функция для триггера DAG и ожидания его завершения
|
|
trigger_and_wait() {
|
|
local dag_id=$1
|
|
warn "Запуск $dag_id через REST API..."
|
|
|
|
# 0. Unpause DAG
|
|
curl -s -X PATCH "$API_URL/$dag_id" \
|
|
--user "$AIRFLOW_USER:$AIRFLOW_PASSWORD" \
|
|
-H "Content-Type: application/json" \
|
|
-d '{"is_paused": false}' > /dev/null
|
|
|
|
# 1. Trigger
|
|
local response
|
|
response=$(curl -s -X POST "$API_URL/$dag_id/dagRuns" \
|
|
--user "$AIRFLOW_USER:$AIRFLOW_PASSWORD" \
|
|
-H "Content-Type: application/json" \
|
|
-d '{}')
|
|
|
|
local run_id
|
|
run_id=$(echo "$response" | grep -o '"dag_run_id": "[^"]*' | cut -d'"' -f4)
|
|
if [ -z "$run_id" ]; then
|
|
echo "Ошибка запуска $dag_id. Ответ API: $response" >&2
|
|
exit 1
|
|
fi
|
|
warn "Успешный триггер. Run ID: $run_id"
|
|
|
|
# 2. Wait
|
|
warn "Ожидание завершения $dag_id..."
|
|
local attempts=0
|
|
local max_attempts=120 # 10 минут (120 * 5 сек)
|
|
|
|
while true; do
|
|
local status
|
|
status=$(curl -s "$API_URL/$dag_id/dagRuns/$run_id" \
|
|
--user "$AIRFLOW_USER:$AIRFLOW_PASSWORD" | grep -o '"state": "[^"]*' | cut -d'"' -f4)
|
|
|
|
if [ "$status" = "success" ]; then
|
|
warn "DAG $dag_id завершен: SUCCESS"
|
|
break
|
|
elif [ "$status" = "failed" ]; then
|
|
echo "DAG $dag_id УПАЛ (FAILED)!" >&2
|
|
exit 1
|
|
elif [ "$status" = "queued" ] || [ "$status" = "running" ] || [ "$status" = "" ]; then
|
|
attempts=$((attempts + 1))
|
|
if [ "$attempts" -ge "$max_attempts" ]; then
|
|
echo "Таймаут ожидания $dag_id!" >&2
|
|
exit 1
|
|
fi
|
|
sleep 5
|
|
else
|
|
echo "Неизвестный статус: $status" >&2
|
|
exit 1
|
|
fi
|
|
done
|
|
}
|
|
|
|
warn "=== DDL: Создание схем и таблиц ==="
|
|
trigger_and_wait "bookings_stg_ddl"
|
|
trigger_and_wait "bookings_ods_ddl"
|
|
trigger_and_wait "bookings_dds_ddl"
|
|
trigger_and_wait "bookings_dm_ddl"
|
|
|
|
warn "=== DAY 1: Initial Load ==="
|
|
trigger_and_wait "bookings_to_gp_stage"
|
|
trigger_and_wait "bookings_to_gp_ods"
|
|
trigger_and_wait "bookings_to_gp_dds"
|
|
trigger_and_wait "bookings_to_gp_dm"
|
|
|
|
warn "=== DAY 2: Increment ==="
|
|
trigger_and_wait "bookings_to_gp_stage"
|
|
trigger_and_wait "bookings_to_gp_ods"
|
|
trigger_and_wait "bookings_to_gp_dds"
|
|
trigger_and_wait "bookings_to_gp_dm"
|
|
|
|
warn "=== Верификация данных ==="
|
|
# Выполняем проверки строк и бизнес-логики, чтобы убедиться, что все витрины DM-слоя заполнены
|
|
docker compose -f docker-compose.yml exec greenplum bash -c "su - gpadmin -c \"psql -d gp_dwh -c \\\"
|
|
SELECT 'STG bookings' as layer, COUNT(*) FROM stg.bookings UNION ALL
|
|
SELECT 'ODS bookings', COUNT(*) FROM ods.bookings UNION ALL
|
|
SELECT 'DDS fact', COUNT(*) FROM dds.fact_flight_sales UNION ALL
|
|
SELECT 'DM sales_report', COUNT(*) FROM dm.sales_report UNION ALL
|
|
SELECT 'DM route_performance', COUNT(*) FROM dm.route_performance UNION ALL
|
|
SELECT 'DM passenger_loyalty', COUNT(*) FROM dm.passenger_loyalty UNION ALL
|
|
SELECT 'DM airport_traffic', COUNT(*) FROM dm.airport_traffic UNION ALL
|
|
SELECT 'DM monthly_overview', COUNT(*) FROM dm.monthly_overview;
|
|
|
|
-- Дополнительная проверка бизнес-логики (load factor не должен быть NULL и должен быть в пределах разумного)
|
|
SELECT
|
|
'Check LF' as check,
|
|
COUNT(*) as total_rows,
|
|
SUM(CASE WHEN avg_load_factor IS NOT NULL AND avg_load_factor >= 0 THEN 1 ELSE 0 END) as valid_lf
|
|
FROM dm.monthly_overview;
|
|
\\\"\""
|
|
|
|
warn "E2E ETL тест успешно завершен!"
|