# Конвеєр ## Що це таке Конвеєр, повніше конвеєр даних, а в професійному мовленні часто пайплайн (від англійського pipeline), — це набір завдань, які виконуються в чітко заданому порядку, причому кожне бере те, що підготувало попереднє, аж поки сирі записи із систем, де вони виникають, не опиняться очищеними, зведеними й збереженими там, звідки їх читають звіти, дашборди чи інші програми. Назва прийшла з оболонки Unix, де символ `|` передає вивід однієї команди на вхід наступної. Більшість конвеєрів працює без нагляду: за розкладом або щойно надходить новий файл. Рідко який конвеєр є прямою лінією. Завданню, що завантажує денні продажі, можуть спершу знадобитися і курси валют, і список клієнтів, тоді як ці два завдання цілком можуть виконуватися паралельно. Завдання разом зі стрілками «має завершитися раніше» між ними утворюють орієнтований ациклічний граф: кожна стрілка вказує в один бік, і жодне завдання не може опинитися в очікуванні самого себе. Apache Airflow, платформа з відкритим кодом для планування й моніторингу таких робочих процесів, називає кожен такий граф Dag — саме так, а не DAG, бо значення цього терміна в Airflow давно вийшло за межі суто математичної структури. Пакетний (batch) конвеєр прокидається за розкладом чи за подією, обробляє обмежений зріз даних, наприклад показання лічильників за один день, і зупиняється. Потоковий (streaming) конвеєр працює безперервно й передає кожну подію далі, щойно вона надходить. Airflow описує себе як платформу для оркестрування пакетних робочих процесів, призначену насамперед для роботи з чітким початком і кінцем, що виконується за розкладом. Найпростіший конвеєр — це рядок у crontab, що о другій ночі запускає скрипт. Цього досить, поки скрипти незалежні. Щойно кілька з них мають іти по черзі, а один падає, комусь доводиться з'ясовувати, що встигло виконатися і що запускати знову. Оркестратор бере цю облікову роботу на себе: він тримає граф, запускає кожне завдання, щойно успішно завершилися ті, від яких воно залежить, повторює завдання, що впали, веде журнал для кожного завдання й кожного запуску і фіксує, які періоди даних охопив кожен запуск. В Airflow кожен запуск за розкладом охоплює інтервал даних (data interval). Щоденний запуск за 1 січня починається одразу після півночі 2 січня, коли день уже минув, і його завдання можуть прочитати межі на кшталт `data_interval_start`, щоб точно знати, який зріз даних їхній. ## Чому це важливо Конвеєр перетворює процедуру на код. Порядок завдань, перевірки між ними, те, що відбувається, коли одне з них падає, і як часто все це запускається, переходять із пам'яті однієї людини у файли, які можна переглядати, зберігати в системі контролю версій і тестувати, як будь-яке інше програмне забезпечення. Airflow описує робочі процеси повністю на Python — підхід, який проєкт називає workflows as code, — і серед переваг називає контроль версій і тестування. Збій до того ж отримує адресу. Коли весь процес — це один довгий скрипт, падіння на сороковій хвилині означає почати з нульової. Якщо ж розбити його на завдання із записаним станом, невдалий запуск показує, яке завдання зупинилося і чому, і перезапускати треба лише його та ті, що йдуть після нього. Airflow дає змогу операторові перезапустити тільки завдання, що впали, або виконати backfill для попереднього періоду, і саме так виправлення, зроблене в середу, доходить до даних за понеділок. Повторення стає безпечним лише тоді, коли кожне завдання для цього написано. Посібник Airflow з найкращих практик вимагає, щоб кожне завдання поводилося як транзакція бази даних, яка ніколи не залишає часткових результатів, і давало однаковий результат за кожного повторного запуску. Таке завдання планувальник може повторити о третій ночі, людина може перезапустити його після виправлення, а за потреби його можна прогнати знову за минулий місяць, і ніхто не мусить спершу перевіряти цільову таблицю. Ціна — ще одне програмне забезпечення, яке треба утримувати. Оркестратор — це окремий сервіс із власною базою даних, оновленнями й правами доступу, а кожен конвеєр у ньому — код, що потребує рев'ю й тестів. Для двох скриптів з одним розкладом чесною відповіддю можуть бути cron і файл журналу. Аргументи на користь оркестратора зростають разом із кількістю завдань, що залежать одне від одного, кількістю людей, яким треба бачити, що виконалося, і ціною ранкового звіту, побудованого на завантаженні, що зупинилося на півдорозі. ## Як це виглядає на практиці [Двадцять GIS-конвеєрів даних і ETL для даних застосунку](/uk/portfolio/automated-delivery-of-20-gis-data-pipelines-and-17/) на одній платформі запускали й відстежували вручну, і весь час існував ризик, що якийсь із них непомітно зупиниться, а дані в подальших кроках виявляться хибними раніше, ніж хтось це помітить. Усі вони перейшли на автоматизоване постачання. Розклад, журналювання й обробку збоїв звели до одного стандарту, тож усі конвеєри поводилися однаково, а той, що зламався, ставав видимим, і їх включили в уже наявну автоматизацію інфраструктури та звітність. Відтоді всі двадцять запускалися самі й поводилися передбачувано, даючи актуальні дані, на які можна покластися, без ручного супроводу. Конвеєр може також збирати щось ціле однією командою. [Генератор CV і портфоліо на основі бази даних](/uk/portfolio/built-a-database-driven-cv-and-portfolio-generator-144/) бере кожен документ з однієї бази даних SQLite на 23 таблиці, а саму базу збирає конвеєр із 19 кроків, що проходить шлях від порожнього файлу до повної бази даних. Кроки мають сталий порядок, і кожен можна безпечно запустити ще раз, тож конвеєр можна проганяти й на вже наявній базі, і жоден рядок при цьому не з'являється двічі. Далі 41 модуль Python загальним обсягом 10 580 рядків створює з цього єдиного джерела шість вихідних форматів, а залежностей під час виконання в нього лише дві. Усередині окремого завдання швидкість важить не менше за порядок. [Конвеєр погодинного агрегування споживання електроенергії](/uk/portfolio/engineered-hourly-electricity-consumption-aggregation-pipeline-in-python-4/) бере дані JSONL, що надходять щогодини, кожні 15 хвилин або в нерегулярні моменти, вирівнює їх за строгими погодинними інтервалами, підсумовує виробництво за міксом, додає імпорт і віднімає експорт, а на виході дає один рядок на годину на зону. Його реалізували трьома способами — на Python, на JQ і на SQL у PostgreSQL — і кожен варіант перевірили на даних за один день і за 30 днів. Найшвидшим виявився Python: 34 мс для одного дня і 180 мс для 30 днів, а його код можна було розширити на більше зон або вбудувати в ETL-конвеєр. ## Де виникають проблеми **Повторний запуск дає іншу відповідь.** Щоденне завдання дописує свої підсумки через `INSERT`, падає на півдорозі, планувальник повторює його, і рядки, записані до збою, тепер лежать двічі. Завдання, що вибирає все від поточного моменту мінус доба, під час повтору читає інший зріз, ніж за першої спроби, а завдання, що читає найновішу версію вхідних даних, підхоплює все, що змінилося в проміжку. Посібник Airflow з найкращих практик називає всі три випадки: замінити `INSERT` на upsert, читати й писати конкретну партицію, прив'язану до `data_interval_start` запуску, і тримати `now()` подалі від будь-якого критичного обчислення. Для таблиці погодинних підсумків у PostgreSQL один простий спосіб — замінювати весь інтервал однією транзакцією. ```sql BEGIN; DELETE FROM hourly_load WHERE hour_start >= :'start' AND hour_start < :'end'; INSERT INTO hourly_load (zone, hour_start, mwh) SELECT zone, date_trunc('hour', read_at), sum(mwh) FROM reading WHERE read_at >= :'start' AND read_at < :'end' GROUP BY zone, date_trunc('hour', read_at); COMMIT; ``` Якщо запускати його як `psql -v ON_ERROR_STOP=1 -v start=... -v end=... -f hourly_load.sql`, а початок і кінець інтервалу даних запуску передає оркестратор, завдання видаляє й переписує рядки рівно цього інтервалу, тож другий запуск залишає таблицю такою самою, як перший. Форма `:'start'` — це власний синтаксис psql для підстановки змінної, взятої в лапки як SQL-літерал. **Завдання завершується зеленим, а дані хибні.** Оркестратор бачить, чи процес завдання завершився чисто, і на цьому його знання закінчуються. psql — показовий приклад: за замовчуванням він продовжує роботу після помилки у скрипті й повертає оболонці 0, тож крок, що виконує `psql -f`, звітує про успіх, навіть коли одна з інструкцій усередині впала. Зі встановленою змінною `ON_ERROR_STOP` psql зупиняється на першій помилці й завершується з кодом 3, і крок падає там, де має. Але й чисте завершення доводить лише, що програма відпрацювала, тому посібник Airflow радить самоперевірки: завдання далі в графі, яке підтверджує, що партицію записано і дані виглядають правильно, перш ніж їх хтось прочитає. Для запуску, що зависає, а не падає, є Deadline Alerts, які з'явилися в Airflow 3.1 на заміну функції SLA, вилученої у 3.0. Вони запускають callback, наприклад лист електронною поштою, коли запуск перевищує часовий поріг. **Повторне ввімкнення конвеєра спричиняє лавину запусків.** З `catchup=True` планувальник створює запуск для кожного інтервалу даних, пропущеного від дати початку, а документація Airflow додає, що catchup спрацьовує й тоді, коли Dag на якийсь час вимкнули, а потім увімкнули знову. Щоденний конвеєр, що місяць простояв на паузі, може повернутися з тридцятьма запусками в черзі до одного й того самого джерела. Значення за замовчуванням, `catchup_by_default=False`, змушує планувальник створювати запуск лише для останнього інтервалу, а документація каже, що Dag, чиї завдання читають поточний час замість свого інтервалу, має тримати catchup вимкненим. Історію краще заповнювати свідомо, командою `airflow backfill create`, яка приймає діапазон дат і обмеження на кількість паралельних запусків через `--max-active-runs`. **Важкий код на початку файлу з Dag гальмує планувальник.** Airflow розбирає кожен файл із Dag знову і знову, щонайменше з інтервалом `min_file_process_interval` секунд між розборами, тож код на верхньому рівні файлу, поза будь-яким завданням, виконується значно частіше за сам конвеєр. Запит до бази даних, мережевий виклик чи повільний імпорт там коштують часу під час кожного розбору, і посібник з найкращих практик наводить приклад, у якому один такий виклик додає до розбору Dag ще 1000 секунд. Залишайте на верхньому рівні лише оголошення завдань і їхнього порядку, а дорогі імпорти й виклики переносьте всередину функцій завдань, яким вони потрібні. ## Що почитати далі - [Apache Airflow](https://airflow.apache.org/docs/apache-airflow/stable/index.html) — огляд від самого проєкту: робочі процеси як код на Python, оркестрування пакетних процесів, backfill і перезапуск лише тих завдань, що впали. - [Best Practices](https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html) — завдання як транзакції, однаковий результат за кожного повторного запуску, код верхнього рівня й самоперевірки. - [Dags](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html) — як Dag оголошує свій розклад, завдання, залежності й типові аргументи на кшталт retries і де місце Deadline Alerts. - [Dag Runs](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html) — інтервали даних, catchup, backfill і очищення завдань, щоб вони виконалися знову. - [Tasks](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/tasks.html) — залежності між завданнями, тайм-аути, політики повторних спроб і перехід від SLA до Deadline Alerts. - [psql](https://www.postgresql.org/docs/current/app-psql.html) — змінні та їх взяття в лапки, `ON_ERROR_STOP` і коди завершення, які читає оркестратор.