Сучасна аналітична модель давно вже вийшла за межі звичайного запиту. Потрібна автоматизація виконання, забезпечення стабільності й однозначності даних, оптимальність швидкості виконання, легкість підтримки і незалежна перегенерація. Ця серія статей написана як певне узагальнення побудови такої моделі без використання високорівневих інструментів, і більше про концепцію й акцентування уваги на базових підходах, ніж про інструменти. Тому якщо потребуєте більше конкретики - будьте вільні написати, я із задоволенням її надам.
Я не відкидаю важливість використання підходящих інструментів і розвитку всієї аналітичної системи, лише більше намагаюся заповнити протилежну чашу вагів, зі своїх спостережень у сфері. Освоїти будь-який інструмент зараз не є складним завданням - велика кількість інформації, що оточує нас забезпечує це. І хайп гонитви у використанні якомога сучасніших інструментів і якомога більшої їх кількості це підживлює. На додачу глобалізація має свої межі оптимальності і не завжди необхідно щось най-най для забезпечення необхідного результату. Тому не про інструменти, а про структурування й розставлення акцентів. А інструменти обирайте самі.
Для ілюстрації підходу тут я використала комбінацію Redshift і Airflow (просто те, що мала в реалізації). Але як було сказано вище, це не про інструменти, а про підхід. Фактично Redshift - це безпосередньо перетворення даних і зберігання, розрахунки і таблиці, Airflow - про кроки, запуск і контроль виконання. Цей самий підхід я реалізовувала на mssql jobs і комбінації Iceberg+Trino+dbt+Airflow, на худий кінець csv файли + скрипт перетворення на будь-якій мові + cron jobs.
І так, вибір української мови свідомий і принциповий. Можливо з часом напишу й іншими, але зараз саме для україномовного середовища. Я роблю це не для хайпу, а для фіксування й поширення ідей, дискусії та взаємодії.
Задача
Побудувати складне багатоетапне перетворення. Під складним маємо на увазі не один селект з віконними функціями й, можливо, декількома вкладеннями, а агрегація різних великих джерел зокрема зі структурними перетвореннями, наприклад з довгого в широкий формат. І це все ще й багатокрокове.
Уточнення: тут йдеться саме про batch (шматкові по часу) перетворення, а не про перетворення на льоту. Для перетворення на льоту важливі принципи, що тут описані, але застосовуються і структуруються зовсім по-іншому.
Критерії:
1) Швидке оновлення. Будь-яке перетворення має бути виконане за раціональний час. Результати за вчора, що розраховуютьс один раз на день вночі можуть в норму рахуватися навіть 10 хв, а от параметри для production що рахуються кожні 10 хвилин бажано вкладати в десятки секунд. У цілому це окрема тема, в той самий час невід’ємний критерій. І якщо він виконаний то покращує результати і інших критеріїв. Тобто швидкість не відкидаємо бо критерії жорстко взаємопов’язані між собою.
2) Підтримуваність у випадку падіння. Щось може працювати без збоїв лише в теорії, тому якщо щось не так маємо відновити роботу процесу в короткі терміни і мінімальними зусиллями. Дуже добре коли це можемо зробити не тільки ми як творці, а і колеги, що додає ваги наступному критерію.
3) Легка модифікація й використання. Модифікація нам потрібна для внесення змін, пов’язаних з новими даними, додатковими критеріями, правилами, особливостями, або ж поруч із поточними розрахунками потрібно розрахувати ще додаткові параметри. Використання має бути легким задля відновлення після падіння, введення в нові процеси, або покращення поточного.
4) Підтримка якості. Найскладніші перетворення ніщо, якщо ми не впевнені в їх результаті. Якщо знаходиться помилка, її потрібно швидко виявити і усунути. Колеги просять уточнити й, як наслідок, поскладнити якийсь висновок - нам потрібно щоб це відбулося без впливу на всі інші показники. І взагалі для того щоб спокійно спати.
Для виконання всіх цих критеріїв нам потрібен читабельний код і структурований алгоритм перетворень. Цей алгоритм - це 5 етапів будь якого перетворення про які піде мова далі. Вони забезпечують модульність, прозорість і втручання тільки в разі необхідності змін і тільки в необхідний шматок.
Етапи перетворення
При роботі з даними розрізняють ETL і ELT підходи. Зараз мова не про них у цілому, а конкретно про частинку T (трансформація) і як її просто забезпечити щоб виконати критерії, зазначені вище.
Усе перетворення можна поділити на наступні кроки:
Перевірка вхідних даних.
Генерація нового шматка моделі.
Критичні тести для нового шматка моделі.
Оновлення базових даних.
Глобальні тести якості.
Кожен з цих кроків є великим, за нього відповідає Airflow і наступний виконується лише у випадку, якщо успішно виконався попередній. Виглядає це наступним чином.
Етапів 5, а тасок в Airflow 4. Пов’язано це з тим що таска merge_task виконує етап перевірки критичних тестів і вставку в загальну таблицю. Тобто можна зробити окремими тасками можна однією, головне щоб в повідомленні про помилку було зрозуміло чому падає. Тут об’єднано, бо merge впаде лише у випадку обриву зв’язку з базою даних, так як дані за межі бази не виходять. Тому якщо падає ця таска то коментар або текст помилки або технічний рядок від Airflow про те що база не відповідає.
Етап 1: Перевірка вхідних даних
Починаємо все з перевірки, бо якщо даних немає або вони не однозначні, то нам немає сенсу взагалі їх перетворювати і запускати наступні кроки. Тут проводимо критичні тести вхідних даних.
Залежно від типу вхідних даних можна виділити наступні 3 сценарії для кожної групи джерел:
Дані ,яким довіряємо: довідники, переливка з інших джерел (зовнішніх або інших команд) можна використовувати безпосередньо, або перевіряти наявність і необхідну мінімальну кількість (відповісти на запитання чи є дані і чи вони потенційно всі).
Дані, які забезпечуються іншими моделями і потоками: перевіряємо чи успішно виконалася попередня модель.
Production дані й дані яким не довіряємо: перевірку відправляємо в окрему модель, яка, окрім перевірки, ще робить фікс того що може і у поточному потоці перевіряємо вже результуючу модель (фактично зводимо до кроку 2).
На зображенні вище checking_dags група із 4 безпосередніх тасок перевірки попередніх моделей - саме моделей, які працюють кожна в своєму незалежному потоці. Перевіряється що вони виконані успішно, а самі дані не перевіряються так як попередні моделі вже покриті тестами і не можуть виконатися успішно якщо критичні тести не пройдені.
Якщо ж в модель входять дані яким довіряємо - тести можна не проводити. Наприклад забираємо дані з GoogleAds де вони детерміновані й однозначні, а їх повноту перевіряє флоу, що переливає. Або ж провести тільки критичні тести на наявність і достатність, як наприклад на наступному зображенні.
Це Airflow таска в якій забираємо необхідний шматок даних і перевіряємо чи взагалі є дані і чи їх кількість є мінімально достатньою. Якщо потрібно провести ще якісь тести то їх легко можна додати наступними кроками, або написати глобальну процедуру перевірки як в етапі 5. І якщо дані потрібно перевіряти глобально, то можливо це не дані яким ми довіряємо? Фактично кожна перевірка - це виконання запиту який повертає кількість рядків (загальну або помилкову) і дивимося на цю кількість. Якщо вона не відповідає критеріям то викликаємо помилку таски (тут функція ValueError у середину якої передається текст помилки), в іншому випадку у разі успішного виконння просто записуємо лог етапу для відслідковування (тут функція logging.info в яку передається текст про те, що саме виконалося).
Чому production дані й дані яким не довіряємо краще відправляти в окремий потік: окрім того, що ви перевірите дані, буде чітке місце де можна сформувати правила виправлення помилок (не тільки вами, а і всіма колегами), і отримати результуючу таблицю з виправленими даними, яким можна буде довіряти, а не дублювати логіку корегувань всюди де використовуються ці дані. Фактично створюється однозначність місця перетворень і даних безпосередньо, тобто повна прозора достовірність. Як бонус можна підкоректувати типи, додати додаткові розрахункові поля, які часто надалі використовуються, оптимізацію тощо.
Відповідно, якщо цей етап не виконується успішно, Airflow надсилає повідомлення про те, що процес впав на конкретному кроці і далі не виконується. Повідомлення про помилки допомагають не перевіряти регулярно, чи все відпрацювало - тобто не заходити вручну, а контролювати тільки одне місце, де зазначено що процес впав і куди потрібно йти за перевіркою і підняттям. При цьому якщо наступний в черзі процес чекає, щоб поточний процес виконався успішно (як на скріні вище), то він буде просто чекати. Ми пофіксимо поточний процес, очистимо Airflow від помилки, перезапустивши конкретний крок виконання, і, якщо все гаразд, то кроки далі в цьому і в наступних процесах виконаються.
Якщо тут не покрити тестами якийсь кейс, то глобальні тести етапу 5 його відловлять. Тобто, на цьому етапі, нам потрібно бути впевненими що ми можемо запускати модель і генерувати дані, а глобальні тести вже покриють все інше. Наприклад, в довіднику маємо порожнє значення, яке не вплине на результат прийняття рішення - то на етапі перевірки вхідних даних можна не формувати тест на порожність, а перевірити її на фінальному етапі і просигналізувати попередній моделі, що вона має неточність в даних (не критичну помилку). Так не перевантажимо поточну модель перевірками.
Етап 2: Генерація нового шматка моделі
З даними все гаразд - впевнилися в цьому - можемо запускати модель і генерувати матеріалізований результат. Не принципово де і яким чином у вас генеруються дані. Принципово, що ми говоримо про модель, а це не просто запит, а їх послідовність з розгалуженнями за необхідності, але, що найважливіше, з вхідними параметрами які дають змогу користуватися моделлю змінюючи лише їх і не лізучи в код безпосередньо. Як функції в програмуванні.
Перш ніж розбирати детально, варто термінологічно розділити 2 таблиці: результуюча таблиця - це таблиця яка генерується безпосередньо цим 2-м етапом, вона має лише дані поточної генерації, а фінальна таблиця - це вже таблиця в якій зберігаються дані за весь час, результати всіх ітерацій генерації.
Зверніть увагу, що тут також, де критично, розглядається різниця між view і процедурою як можливими обгортками перетворення.
Отже, цей модельний блок має відповідати наступним критеріям:
1) Оптимальний час виконання.
Про це вже писала вище, але трохи повторюся: залежно від того як часто і на якому об’ємі запускається модель вона має зрозумілий час виконання і об’єм вхідних даних; в залежності від складності - кроки виконання; в залежності від dwh - особливості виконання функцій, запитів і оптимізація плану виконання запиту. Якщо вдається виконати цей критерій, то і сама модель в регулярному варіанті буде працювати стабільно і добре, і перегенерація у випадку помилки або падіння займе раціональну кількість часу, і якщо в оптимальність ще закласти структурність - то підтримувати і модифіковувати також буде легше.
Тут є один важливий аспект: якщо модель у вас загорнута у view - слідкуйте за “прямим доступом до початкового часового стовпчика”. Що мається на увазі? Для того щоб матеріалізувати view ви будете писати шматок на зразок:
INSERT INTO назва_результуючої_таблиці
SELECT *
FROM view_name
WHERE datetime >= час_з_якого_починаємо_рахуватиТо, для того щоб не опрацьовуват всі дані, а потім тільки вибрати необхідні кілька днів доречно щоб стовбець datetime, який є результуючим у view, напряму брався з початкової таблиці, на базі якої будується view. Тобто якщо вдасться написати щось на кшталт наступного:
CREATE VIEW AS view_name
WITH base_table AS (SELECT datetime, інші_необхідні_стовпчики_таблиці
FROM перша_таблиця_даних),
second_table AS (SELECT b.*, інші_необхідні_стовпчики_другої_таблиці
FROM base_table AS b
LEFT JOIN друга_таблиця_даних)
SELECT *
FROM second_tableто стовпець datetime напряму забирається і в явному вигляді доходить до результату - так вибірка з view буде оптимальною, бо одразу забереться тільки необхідний шматок. Якщо так не вдається,то можливо краще не використовувати в’ю, бо ймовірність, що виконається весь об’єм, а тільки потім відпрацює фільтрування по даті високе. А це і час і операційні ресурси, які немає сенсу використовувати коли результат вже відомий з попередніх ітерацій.
2) Часовий інтервал - вхідний параметр.
Так як говоримо про batch processing в нас часовий інтервал може змінюватися. В регулярній роботі генеруємо дані за останні 2 дні, 10-го числа місяця перераховуємо весь попередній, фіналізуючи дані (на випадок якщо щось долетіло), і перегенерувати з якогось числа, бо продукт отримав нову фічу, її додали до моделі, і є сенс перерахувати тільки дані після її додавання + ситуації падіння в довільний день. Для view - це буде обмеження у WHERE, для процедур - вхідний параметр що передається в середину.
Фактично вхідний інтервал потрібен для швидкого використання моделі без копіпаста і підтримки різних версій + якщо колеги хочуть собі генерувати окремо віддаємо їм документацію з назвою (що викликати) і описом вхідних параметрів. І вони спокійно користуються. Маємо єдиний підхід до створення показника.
3) Модель не тільки заміняє дані в результуючій таблиці, а й може створити її за необхідності. Результуюча таблиця теж вхідний параметр.
Тут треба пояснити трохи детальніше. Це зручність! Якщо ви використовуєте view, тобто ваш запит повердає дані, то зовні загорнути це в створення таблиці або вставлення даних у неї не створює складності.
На противагу, процедури не повертають дані - це виключно набір виконуваних команд, і якщо вам потрібно створити таблицю, то ви окремо можете писати команду створення прописуючи назви стовпчиків і їх типи, або передати параметр action (яку дію виконати). Можна видалити і вставити дані (повне оновлення існуючої таблиці), можна просто вставити (insert або merge, хоча в рамках парадигми це не дуже цікаво), але можна створити. І в цьому випадку модель при відповідному переданому параметрі просто створить таблицю з уже прописаною всередині неї структурою і типами (яке забезпечується наступним пунктом). Можна звичайно скопіпастити, але можна просто передбачити параметром.
На додачу це дозволяє при передачі колегам використати цю можливість і не лізти в код - не давати їм доступу без необхідності.
Результуючу таблицю зручно передавати у вигляді параметра для відслідковування того, що ми створюємо або оновлюємо потрібну нам таблицю - і водночас різні команди можуть користуватися однією процедурою і генерувати незалежні таблиці. Коли потрібно внести модифікації, то модифікації вносимо в код, і передаємо нову назву таблиці з відповідним параметром action що не зачіпає основний процес і в момент готовності дозволяє легко підмінити процедури й результати. Привіт, функціональне програмування.
4) Жорстка фіксація структури і типів. Ця частина допомагає створювати однозначні результуючі таблиці - скільки б їх не було. Але найважливіше (особливо якщо у вас не жорсткотипізований dwh) це сконвертувати все в єдиний формат і не залишати цей формат на вибір бази або сховища, бо він не знає ваші дані. Для прикладу в Redshift рядок зазвичай має максимальну розмірність 65к знаків, а розмір одягу займає максимум 6 символів (це якщо ви пишете XXXXXL, а не 5XL - бо тоді 3). В реальності в Redshift жорстка фіксація типів на одній з таблиць зекономила нам близько 20% пам’яті - тобто жорсткотипізована та ж сама таблиця важила на 20% менше. Крім тоо це впливає не лише на зберігання, а й на подальше використання цих даних бо їхній обсяг просто фізично менший. І виділення цього перетворення як окремого шматка дозволяє відразу сказати, що в нас прийшли якісь неконсистентні дані. Згадуємо про початкові перевірки і що цей кейс напевно не був покритий, текст помилки нам вказує який стовпець не зміг сконвертуватися відповідно → йдемо в початкову таблицю звідки забираємо цей стовпчик → знаходимо відповідні рядки → визначаємо як і на якому етапі фіксимо + маякуємо колегам що формують ці дані → перегенеровуємо відповідний шматок (можна так само як і з помилкою) і весь процес рухається далі.
5) Врахування інших параметрів за необхідності. З view це складніше, а на прикладі процедур досить просто. У випадку, якщо вам потрібно використовувати дані з різних таблиць - можна передавати назву цієї таблиці, а не підтримувати 2 процедури. Якщо потрібно оброблювати в різний час різні типи даних - можна задати передавання цього типу, а в процедурі описати безпосередньо цю обробку. Реальний кейс, який ми вирішили таким чином - це “долітання чеків”. Дані перетворень в нас не змінювалися бо це були логи, а от чеки могли прийти пізніше. найяскравіший приклад - це чек повернення так як товар можна повернути протягом 14 днів з моменту покупки. Тому процедура отримувала вхідний параметр cheque_type і в залежно від того, які чеки ми хотіли додати: нові, повернуті чи обидві групи формувався хвостик WHERE відповідним чином.
На двох зображеннях нижче показаний початок і кінець процедури яка відповідає і ілюструє все що стосується цього етапу.
У процедуру передаємо:
часовий інтервал розрахунків start_datetime, end_datetime щоб у будь-який момент згенерувати будь-який проміжок часу (до речі це допомагає знайти долетівші дані також, бо зараз даних більше ніж тоді коли процедура штатно генерувала дані за розкладом).
rez_table - назва результуючої таблиці в яку запишемо результат.
action, про який явно говорилося вище - дія, яку потрібно зробити з результуючою таблицею.
cheque_type - тип чеків, що додаємо.
Тоді на початку визначаємо команди для дій з результуючою таблицею і хвостик для додаткової вибірки.
В кінці процедури проводимо жорстку типізацію і залежно від переданого параметру action виконуємо відповідну дію з результуючою таблицею (зокрема і створення).
Як наслідок, щоб вам не знадобилося зробити поза основним data flow вам потрібен тільки 1 рядочок для отримання результатів.
Увага! Результатом цього етапу є результуюча таблиця, а не фінальна. Ми записуємо результат перетворення і йдемо до наступного етапу - перевіряти критичні властивості моделі.
Етап 3: Критичні тести для нового шматка
Критичними визначаємо властивості, які надалі призведуть до бізнес-вагомих некоректних висновків.
Критичних тестів зазвичай небагато, але вони мають відповідати на запитання “Чи можна далі користуватися даними?”. Зазвичай для цього нам достатньо перевірити що дані взагалі згенерувалися і що вони є гранулярними. Залежно від моделі можуть бути ще інші перевірки, наприклад відсутність порожніх значень тощо. Досить просто коли маємо окремо згенеровану результуючу таблицю:
- Запитали скільки даних, порівняли їх з необхідним значенням. Якщо мало - звалили з помилкою недостатньо даних.
- Запитали про кількість рядків більше одного для ключа, порівняли з нулем. Якщо не нуль - звалили з помилкою про негранулярність.
- Потрібно щоб критичний стовпчик відповідав параметрам. Запитали скільки рядків не відповідає параметрам. Не нуль - звалили з відповідним текстом помилки.
- Потрібно щоб загальна сума відповідала загальній сумі вхідної таблиці, перевірили. Якщо не відповідає - звалили з відповідною помилкою.
І так по всім критичним.
Якщо в нас процес впав на цьому етапі - ідемо в Airflow, дивимося помилку, копіюємо текст селекта з результуючої таблиці рядочків що є помилковими і відразу бачимо причину помилки не перегенеровуючи нічого і не чекаючи 5-25 хв доки модель згенерує дані. А якщо дані таки долетіли, або колеги їх встигли вже підкорегувати - то відтворити помилку без результуючої таблиці взагалі буде неможливо. Приклад такої перевірки відразу з наступним етапом вставки коректних даних в фінальну таблицю.
Етап 4: Оновлення базових даних
Показаний на зображенні вище. Фактично якщо критичні тести пройдені - можемо вставляти в фінальну таблицю де зберігаються всі дані (а не один часовий шматок). Нічого складного окрім стратегії. Умовно їх дві видалити і вставити, або оновити:
1) DELETE+INSERT вимагає жорсткого відстеження дат або інтервалів тому тільки з наскрізними датами (коли для генерації використовується той самий інтервал що і в результатах).
2) MERGE/UPSERT оновлює лише нові згенеровані дані без жорсткого відслідковування але для цього потрібна унікальність і якщо було згенеровано щось зайве то перезапуск всього процесу не допоможе.
І тут критично розуміти результат і як він пов’язаний з початковими даними. Якщо потрібно нічого не втратити - дані оновлюємо, якщо потрібна відповідність входу і виходу то видаляємо і вставляємо. Найяскравіше різницю можна показати на наступному прикладі: домовилися, що у вхідних даних не повинно бути тестових, але колеги таки знайшли тестовий рядок і видалили його. Тоді у вас теж має його не стати і тут краще працює DELETE+INSERT. А якщо ви маєте показати всі рядки незважаючи на те вони видалилися в початковому джерелі чи ні (максимум промаркувавши їх) - то однозначно ваша стратегія MERGE/UPSERT.
Періодично з часом ці стратегії доводиться змінювати (наприклад тестів стало дуже багато і домовилися їх виключати, або взагалі бізнес процес змінився і немає прямої зв’язки вхідного часового інтервалу і результуючого), і так як результуючу таблицю ми маємо нам потрібно просто змінити текст одного запиту не чіпаючи жодне інше місце всього процесу.
Етап 5: Глобальні тести
З моделлю в нас все добре, результати ми отримали і можемо з ними працювати. Тож, як вишенька на торті, перевіримо чи все в нас нормально, а не лише мінімально необхідні перевірки. Критичні тести призведуть до поганих наступних результатів, глобальні тести - про якість даних і результатів: наприклад колеги додали нову фічу і в категоріальному стовпчику маємо новий параметр, або фічу додали, про нову категорію ми знаємо але забули показник передавати.
Перевіряємо результуючу таблицю - бо немає сенсу перевіряти увесь об’єм - треба тільки те що згенерували в поточній ітерації. На додачу одна помилка не є інформативною - доречно знайти всі для чистоти моделі. А обмеження накладаємо на стовпчики. Тобто просто запитуємо з результуючої таблиці рядки, які не відповідають правилам для кожного стовпчика. Наприклад нижче ідентифікатор замовлення порожній
І використовуючи UNION ALL об’єднуємо ці рядки в єдину таблицю помилок, додаючи стовпчики з типом помилки і датою перевірки. Зручно, коли цей великий запит перевірки загорнутий у процедуру, вхідним параметром до якої є назва результуючої таблиці, щоб можна було перевірити будь-яку таблицю згенеровану моделлю.
Якщо в результаті в таблиці помилок маємо хоча б один рядок за сьогодні (або за останню ітерацію перевірки) звалюємо Airflow таску наступним чином
Коли бачимо це падіння можемо зайти в таблицю помилок і подивитися, які рядки і чому впали без повторної генерації всього.
Висновок
Всі ці 5 етапів дозволяють прозоро і без зусиль виконати всі критерії, про які говорили вище:
Швидке оновлення забезпечується швидкістю генерації моделі і відсутністю непотрібних перегенерацій.
Підтримуваність у випадку падіння - чітко знаємо на якому етапі впала послідовність і чому (бо передаємо текст помилки), не перегенеровуючи, можемо побачити помилку і як наслідок без зайвих дій пофіксити її (бо час витрачається тільки на фікс, а не на пошук).
Легка модифікація і використання - завдяки обгортці генерації і перевірки в суцільні блоки з вхідними параметрами, використання зводиться просто до виклику в один рядок, а модифікація відбувається тільки в необхідному блоці без впливу на інші і з необхідними перевірками.
Підтримка якості - так як маємо критичні і глобальні тести, як тільки щось не так отримуємо сигнал від процесу й приймаємо рішення про фікс, а не до нас через деякий час приходять із повідомленням, що знайшли помилку і починається глобальне дослідження “що ж сталося”.
Наслідок для технічної команди:
зосередженість на нових задачах і питаннях
якщо все працює штатно, працюємо тільки з помилками - прозоро і швидко
легко внести зміни за необхідності, або поділитися генерацією підтримуючи єдиність формування показників
спимо спокійно
Наслідок для бізнесу:
даним довіряємо в 99% випадків, в 1% технічна команда прийде з повідомленням що все пофіксили в короткі терміни
впевнено приймаємо рішення і використовуємо їх
спимо спокійно
Сучасні технічні засоби - чудово, прозора логіка і структурованість - необхідна, в комбінації отримуємо “золотий грааль”. Якщо тільки прозора логіка і структурованість - отримуємо лише результат. Якщо тільки сучасні технічні засоби - створюємо “скриньку пандори” з якою не будуть знати що робити і боятися відкривати. Тому тільки комбінація!
Якщо щось із описаного вище викликає у вас запитання чи коментарі - буду рада їх почути/читати тут або в особистих повідомленнях. Відкриті дискусії вітаються. Критика теж.
