Логи! Если хочешь потратить время

Иногда у инженеров или аналитиков возникает идея извлечь полезные данные из логов. Иногда даже разработчики приложений сами говорят нам: ну вот же логи! там есть данные.

Работая с операционной базой, мы уже готовимся к неожиданностям, которых вроде бы должно быть немного, ведь работа приложения описана и тестируется. А логи...

Логи существуют для дебага работы приложения. Реже для целей мониторинга и алертов.
Таким образом, логи - технические данные для разработчиков и причины их изменения могут быть самые непредсказуемые.
На логи:
❌НЕ пишут бизнес требования
❌НЕ пишут документацию
❌НЕ делают контракты на пейлоад
❌НЕ проверяют качество и не тестируют.

а если все это делают то это уже не логи

Туда разве что для ресерча можно залезть👀 (хотя для ресерча все можно) и то я бы полез туда только в крайнем случае: оставим дебаг тому кто это наделал разработчикам приложений.

©️Извлечение аналитических данных из логов - ступенька вниз к хаосу или шаг в болото - пока по колено зашел вроде ниче, а чуть расслабился и затянуло с головой и тратишь все свое время на поддержку

Особенности foreign data wrapper

Ранее я писал о плюсах использования внешних таблиц, теперь хочу написать об особенностях, которые в нашем use case приводят к проблемам. Спойлер: мы отказались от использования внешних таблиц.🙁

У foreign data wrapper есть особенность:
Fdw запрашивает фул таблицу с удалённого сервера, даже если вы применили фильтры в запросе.😮 Очевидно, что если внешняя таблица на миллионы строк, то ждать придётся очень долго пока вы получите всю таблицу. Эту особенность можно обойти, если “помочь” fdw понять как правильно подготовить запрос ко внешней таблице. Вот здесь хорошо описано и приведены способы написания запроса оптимального для внешних таблиц.🙌 Поэтому я назвал это особенностью. Сама по себе особенность не проблема, работать можно.☝️

А проблема обусловлена именно нашим кейсом.👀

📝 Ситуация (Кейс):
У нас есть DWH на Postgresql и есть Data Marts тоже на Postgresql. Дата марты содержат сабсеты данных специфические для какой-то задачи/домена/департамента. В тех случаях, когда данные в дата марте полностью совпадают с данными в двх как раз возникает отличная возможность избавиться от лишних ETL процессов. Тут то мы и применили внешние таблицы.
В нашем кейсе дата март - конечный продукт для пользователей. Туда приходят делать запросы для анализа, ресерча, построения отчётов и тд. Дата март как продукт должен предоставить возможность работать с данными, делать простые или сложные запросы, оптимальные или неоптимальные.

📝 Проблема:
Пользователям все равно какие там технологии применили под капотом дата инженеры. Естественно они не должны держать в голове, что там какие-то внешние таблицы и как-то правильно надо писать запросы, более того они вообще не должны знать о таких подробностях.☝️ В итоге это привело к резкому падению скорости обработки запросов, жалобам пользователей и отказу от использования внешних таблиц в этом кейсе.💔

📝 Решение:
Физическое хранение данных в самом дата март. Мы перешли к плану Б и настроили логическую репликацию на те таблицы, схема которых совпадает в двх и дата март, то есть там же где были внешние таблицы. 🙌

©️Но мы не убираем foreign data wrapper “на помойку”! В других кейсах, где такими внешними таблицам будут пользоваться инженеры можно работать эффективно. Так что отложили пока в сторонку.
Кстати mysql_fdw тоже попробовал и он тоже отлично работает и настраивается так же легко как и postgres_fdw.

Занимайтесь творчеством!

Когда я пробовал генерировать картинки с AI с меня слетели розовые очки.🫣 Я почему-то думал, что я дам ему текст поста и он что-то интересное нарисует. Как я ошибался! Ему нужно точное описание что нужно сделать: стиль композиция, фон, цвета, детали, эмоции, элементы, формы и все все. Получается нужно в голове полностью придумать изображение и описать его! Творческий процесс! И тут я наткнулся на мысль, а ведь творчество - это про придумать, а не про сделать.

Я порассуждал на эту тему и склоняюсь к тому, что продумать само решение - вот творчество! И с этим AI пока не справляется. Он может заменить последний этап, этап производства.

На самом деле так было всегда. 🤔
💃 Дизайнер одежды не шьёт, а сначала рисует, потом берет разные лоскуты и иголками или ещё как-то закрепляет на манекене, что-то меняет, думает, а готовый вариант может сшить подмастерье.
🏡 Дизайнер (домов, интерьера, рекламы, машин и т.д.) тоже продумывает композицию стиль, потом делает эскизы, рисует, а затем в заданном стиле можно создать множество изображений и добавить деталей.
🎨 Художник тоже изобретает стиль и какой-то уникальный подход. Знаменитые художники знамениты за их "новинки", уникальность, которую они придумали, а репродукцию (копию картины) могут сделать сотни или тысячи художников в мире. Насколько я знаю (где-то читал), у знаменитых художников были подмастерья, которые помогали писать, добавляли какие-то простые делали, фон типа неба или земли - делали рутину. Сегодня это может делать AI.
🎹 Композитор не играет сам партию каждого инструмента, он придумывает главную тему используя как правило только пианино, описывает ее, а потом ее можно масштабировать меняя оркестровку, темп, инструменты и тд (послушайте Ханса Циммера и его саундртеки к фильмам).

👨‍💻У нас как инженеров тоже такой процесс. У нас есть творчество! Придумать решение - творчество, написать код - рутина. Больше занимайтесь творчеством! Опустите технические детали до того момента, когда будете писать код. Опишите проблему и решение естественным языком. Нарисуйте ваше решение с простыми квадратиками и стрелочками (инструменты типа uml придут позже, когда вы сами поймете что они вам нужны), затем уже опишите конкретные шаги (пост про псевдокод). Решение готово! Кода нет? Напишите сами или отдайте AI - это рутина. ☝️

©️Занимайтесь творчеством ребята!

Ситуация©️ Проблема ©️Решение

Большая часть работы состоит из решения проблем и различных коммуникаций по поводу их решения. Часто нужно описать проблему, поставить задачу так, чтобы твой коллега все понял. А чтобы объяснить кому-то нужно ещё понять проблему самому, добраться до сути. 🧐

Есть отличный и очень простой фреймворк, который помогает структурировать мысли, выделить суть и логически выстроить информацию, чтобы лучше понять проблему самому и доступно и убедительно объяснить её любому. 🙌

📝 Ситуация-Проблема-Решение или The "Situation-Complication-Resolution" (SCR) фреймворк широко используется в бизнес коммуникациях, сторителлинге, решении проблем.

🔸 Ситуация: основное на фактах описание текущей ситуации или состояния, окружения, контекста
🔸 Проблема: конфликт или разница между текущим состоянием и желаемым, блокер, то что требует действий.
🔸 Решение: что нужно сделать, чтобы разрешить проблему

По сути это метод рассуждения, который можно применять везде, а рассуждать я люблю и умею🤓. Я им пользуюсь часто для сложных и простых ситуаций, для постановки задач, для переписки в рабочих чатах или каналах и даже вот применил в посте. При этом не обязательно описывать (проговаривать) каждый из пунктов явно, можно о них подумать, а описать уже просто в виде нескольких связанных предложений.

И самое лучшее во фреймворке то, что применять его очень легко! Можете попробовать в понедельник описать таск используя СПР (SCR) фреймворк.

Кстати у нас в компании head of engineering так любит этот фреймворк, что у нас тикеты на команду девопсов имеют обязательные поля для заполнения Situation, Complication, Resolution.☝️

Пример есть в посте и еще пример в комментариях для простой технической задачи из недавних 👇

Отношение меняет все

Как бы по инфоцыгански это не звучало, это так. Наши установки меняют мышление и настрой.

Плохие установки приводят к проблемам в коммуникации, к проблемам в доведении проекта до результата и тд.

Вот плохие установки, которые часто встречаются. И у меня тоже🙁, но я с ними борюсь

❌Легаси код - плохой код. Есть такая цитата: легаси - это снизсходительное название того, что прямо сейчас приносит деньги.
❌Заказчик/клиент глупый и не знает чего хочет. Классика. Лучше исходить из того, что заказчик - эксперт в своей области и нужно его понять и направить в нужное русло.
❌ Хочешь сделать хорошо, сделай сам. А потом выгори, сорви сроки, оттолкни от себя людей, которые хотят и могут помочь.
❌ Это не моя проблема, это их проблема. Все проблемы - это проблемы бизнеса. А мы работаем на бизнес. Это не значит что все нужно делать самому (смотри предыдущий пункт), но подсветить проблему, поделиться мнением, создать таск в беклог 😁.
❌ Правила мешают работать. Однако их придумали неспроста. Они написаны кровью, пОтом и матами инженеров, которые были вынуждены разгребать хаос.
❌ Безопасники мешают работать (долго доступ дают и вопросы какие глупые задают). Во-первых смотри предыдущий пункт. Во-вторых тема безопасности данных вообще очень важна для нас и тут, наоборот, мы, как дата инженеры, должны первые встать и сказать, что с безопасностью что-то не так. Кстати в Fundamentals of Data Engineering безопасность данных - часть работы дата инженеров и их ответственность тоже .

Modern data stack

Цитата из книги Fundamentals of Data Engineering о современном дата стеке.

the main objective of the modern data stack is to use cloud-based, plug-and-play, easy-to-use, off-the-shelf components to create a modular and cost-effective data architecture.

Меня позабавило скопление в одном предложении всех этих словосочетаний подчеркивающих простоту и быстроту развертывания и использования 😄. Может даже показаться что и дата инженеры-то не нужны😯. Но нам не о чем беспокоиться! Даже при использовании таких компонентов остаётся сложность создания собственно архитектуры с правильными компонентами и с правильным взаимодействием. Нужно стремиться стать гибким инженером (это я самому себе 😉), способным эффективно использовать модерн дата стек, не застревая в двух крайностях:

😡Олдскул: берём только проверенные временем компоненты, а все чего не хватает пишем сами.

😡 Resume-driven development - выбор исключительно крутых новейших технологий, забывая конечные цели проекта. Вот ещё одна цитата из той же книги.

There’s also a temptation to do resume-driven development, stacking up impressive new technologies without prioritizing the project’s ultimate goals.

©️Перечитал пост и понял, что как и всегда все снова сводится к балансу и здравому смыслу 🤔

Мониторинг и алерты

Я очень люблю эту тему и почему-то о ней несправедливо мало говорят в контексте дата инжиниринга и вообще работы с данными. 

📝 Я убежден, что мониторинг и алерты неотъемлемая часть хорошего инженерного решения.

Мы стараемся закрыть мониторингом и алертами все, за что отвечает наша команда. Иногда ребята меня спрашивают зачем нам приоритезировать работы по настройке мониторинга и алертов. Я обычно отвечаю, что задача алертинга уведомить нас о проблеме до того, как о ней узнает бизнес. Алерты работают на качество нашего сервиса (даже если вы работаете всегда внутри своей компании, заказчики - это аналитики или стейкхолдеры и вы оказываете им услуги). Все это правильно, но я чувствовал что тут есть что-то ещё, что меня беспокоит, когда нет алертов. И вот мне наконец удалось порассуждать, добраться до сути и сформулировать ее 🙌.

Пассивная уверенность vs активная уверенность.

Пассивная уверенность - состояние, при котором в любой момент времени, не предпренимая никаких действий, точно известно, что все работает хорошо. Например CEO подошел и спросил "все работает?" и ты через секунду уверенно отвечаешь - да.

Активная уверенность - состояние при котором, чтобы убедиться что все работает хорошо, нужно предпринять какие-то действия: открыть ui оркестратора, проверить наличие ошибок в логах и тд.

Вот оно! Я понял, что меня психологически не устраивает активная увереность. Поэтому я стремлюсь к состоянию пассивной уверенности.

Ксати я раньше не встречал нигде такие понятия и определения. Возможно я даже что-то новое сформулировал😎. 

Кроме психологической причины есть ещё конечно чисто практическая. Пассивную уверенность можно масштабировать на неограниченное количество контролируемых единиц (сервисов, приложений, функций, дагов, етл процессов) - добавление очередной единицы не влечёт временных затрат на проверку. (только время реакции на алерт если он будет). А вот активная напротив плохо масштабируется по понятными причинам: новая единица - время на проверку.

Лучший друг человека

Вчера весь день работал с другом. Я изучал работу со spark на aws emr serverless, aws glue, data catalog, а он мне помогал. Я говорю дай скрипт, который будет делать вот это, он даёт, я говорю, а почему вот тут не сделал проще? Он: точно! и поправляет. Я: вот тут какие ещё варианты? Уточни как это работает. А что под капотом по факту происходит? А вот тут похоже будут проблемы в таких-то кейсах? И тд...

Наверное вы уже догадались, что это ai chat. В этот раз конкретно я общался с windows copilot, а вообще я и с другими друзьями общаюсь 😂: с chatgpt, amazon q (когда в aws console), githab copilot (когда работаю в ide). Эта компания мне здорово помогает! 👍 Но у этих ребят конечно надо знать, что спрашивать! И я даже не про промпты, а про то, что нужно понимать контекст и главное понимать свою цель. ☝️

Раньше процесс изучения новых для меня технологий я начинал с обзорных статей, потом к статьям уже более детальным - туториалам, потом уже к документации или к еще более глубоким статьям. Теперь я заметил, что следующий шаг после обзора хорошо заменяют такие чаты, так как дают быстрый доступ к нужной информации (что логично ведь на основе открытой информации в интернете они и обучены). Таким образом сначала все таки статьи, чтобы знать что к чему и знать, что спрашивать у друга. Потом ai друг. Ну а глубже пока эти друзья не тянут. Не то чтобы ошибаются, но скорее привирают или обобщают/упрощают. Для начального этапа погружения это ок, но для решения сложных проблем уже не то.

В итоге мой план на любой новый инструмент:
1️⃣несколько обзорных статей наискосок,
2️⃣пообщаться с ai другом и проба что-то сделать вместе с ним,
3️⃣сделать рабочее решение моей задачи
4️⃣уточнить детали в документации по мере необходимости.
5️⃣А уж если цель - экспертность, тогда книга.

В прошлом посте я упомянул, что изучал некоторые инструменты на aws для работы с данными. За один день я потрогал AWS EMR, AWS EMR Serverless, AWS Data Glue (catalog, crawler).

Как и всегда изначально у меня была четко сформулированная проблема, а точнее две. Это потом я уже выдвинул гипотезу, что эти проблемы можно решить на aws и потому туда и пошёл. Просто я свое время ценю и инвестирую его осознанно 😉

📝 Итак проблемы:
1. Compute выполняется на воркерах airflow, что очевидно ограничивает работу с большими данными.
Гипотеза: перенести работу в spark, а aws emr упростит поддержку кластера.
2. Зависимость от Postgresql. У нас DWH на Postgresql и архитектурно это ограничивает свободу выбора инструментов работы с данными.
Гипотеза: отделить слои трансформации+хранения от репрезантативного слоя, организовать data lakehouse с данными в открытом формате (iceberg) на aws s3, а postgresql сделать опцией.

✏️ Далее мои выводы и впечатления. Вкратце в виде обзора. Для тех кто уже знает эти инструменты ничего нового не скажу.

Пару слов о spark. Я думаю все как минимум слышали о нем. Это compute engine для распределенной обработки данных. Он изолирует все сложности под капотом, а снаружи дает простой интерфейс для разных языков программирования. Ему нужен кластер.

Aws emr как раз дает кластер, но он хоть и облачный, но все равно требует поддержки (конфигурация, мониторинг, деплой и тд), а мы этого делать не хотим, нам же надо modern data stack, значит сразу идем смотреть serverless. Тут aws даёт возможность забыть о кластере и просто слать ему job 👍. Aws при этом предоставляет библиотеку для airflow, в которой есть операторы для работы с Aws emr serverles. Супер! Все что нужно - это выполнить submit job!

В нашем случае мы уже изолировали оркестратор и даги строятся автоматически из конфигов и все что нужно сделать - это подготовить универсальный spark job script и подложить его в билдер. И вуаля! Десятки наших дагов начнут работать на spark и даже об этом не узнают 😁 - решение проблемы 1️⃣. А так же здесь можно перенаправить поток данных в aws s3 в файлы iceberg, так как spark умеет с ними работать - решение проблемы 2️⃣. В общем-то уже тут я понял что не зря пришёл и с этим можно работать. 🙌

Продолжение в следующем посте ⬇️

Продолжение предыдущего поста

AWS Glue даёт возможность изолировать зоопарк ваших дата источников/дестинейшнов сколько бы их ни было. Базы данных, event stream (kafka), файлы любых форматов в aws s3 или где-то ещё. Он все это может понять и позволить работать со всем этим взаимодействуя только с одним интерфейсом, одной точкой - aws glue.

Это возможно при создании дата каталога - такое хранилище мета данных или другими словами схем данных: какие таблички в базе да с какими колонками и тд. При этом есть специальный инструмент - crawler, который по запросу или расписанию ходит везде и обновляет мета данные ⛏ . Таким образом aws glue знает, что вот тут postgresql база и в ней такие таблицы и колонки, а вот тут mysql, а вот ещё kafka и тд. Это и позволяет Data Glue сидя сверху на каталоге делать грязную работу работать со всем многообразием за вас. И круто ещё и то, что под капотом spark! Все таблички из каталога уже в контексте! Вы в скрипте обращайтесь к чему угодно как к таблице, а дальше уже происходит магия и данные подтянутся будь-то csv файл, будь-то mysql табличка! (Сам на проде не пробовал пока, интересно как он будет билдить запросы. Поглядим 🤔).

В нашем случае ещё сильнее упрощается скрипт и упрощается работа с sql, так как не нужно думать о подключениях, базах, отличиях синтаксиса и тд. Звучит круто! Потом попробую на проде, расскажу 😄 У aws glue ещё есть свой etl оркестратор типа для визуального построения флоу, но он что-то с первого взгляда меня не впечатлил.

Ну и вишенкой на торте на дата каталоге лежит AWS Athena. Если Glue даёт возможность работать со спарк скриптами, то этот инструмент даёт обычный sql интерфейс над всем каталогом 💪. Можно подключиться через ваш любимый database клиент, типа dbeaver или datagrip и работать с sql консолью, не думая о том, что там могут быть десятки баз данных и файлов! 😮 (тут я немного переживаю, не получится ли как с foreign data wrapper и его далекими от оптимальности запросами, пробрасываемыми непосредтсвенно в базу-источник, но есть надежда что в aws сделали получше)

Должен ещё упомянуть, что все это добро стоит недешево. Порядок чисел начинается с сотен долларов на минималках. Если мы решим туда идти, то обязательно будем внимательно следить за стоимостью и переходить постепенно. К слову есть такая штука как FinOps (что? Уже есть devops, dataops, а это ещё что? 😳). Да, оказывается дата инженеры ещё и немного финопсы! (Книга Fundamentals of data engineering). За modern data stack со всеми этими off-the-shelf, easy-to-use, plug-and-play надо платить. 💰

©️В итоге я понял, что обозначенные мной проблемы можно решить красиво, молодежно в AWS.

Спасибо, что прочитали этот длинный двойной пост 🤗

В выигрыше тот, кто делится

Нужно делиться экспертизой. И не потому что это полезно для вашей команды с точки зрения развития младших товарищей или с точки зрения исключения ситуаций, когда кто-то один обладает уникальными знаниями о реализации чего-то.

В этом посте я хочу сказать, что делиться экспертизой полезно именно для каждого отдельно взятого индивидуума.

При этом я знаю, что есть люди, которые намеренно не делятся знаниями, думая, что таким образом делают себя уникальными и незаменимыми! 😳 Не понимаю таких и их даже немного жалко: знания устаревают, а они топчатся на месте.

под спойлерами ниже вы найдёте параллели с базами данных, которы я добавил по приколу 😁
Я понял, что когда объясняшь кому-то что-то, то как бы:
©️Лучше структурируешь информацию в голове у себя самого
. (Индексация, метаданные, связи)
©️Фиксируешь знания и освобождаешь пространство для новых. Не знаю как это работает, может типа в долговременную память уходят.
(Партицирование, сохранение на диск и освобождение кэша)
©️Освобождаешь время за счёт делегирования задач, где больше нет неопределенности - рутина.
(Тут не придумал аналогию. Может индекс скан?)

В итоге в выигрыше тот, кто делится!

Я и канал этот создал, потому что хотел поделиться с вами кучей мыслей, которые уже скопились в заметках и ещё возникают каждый день 😃

А канал мне помог создать Влад и его курс по ведению тг канала. У меня было много сомнений и пугала неизвестность, так как я ничего не знал о ведении канала. Влад помог и сомнения развеять и лайфхаков насыпал и нужный настрой задал! Ещё круто, что с первого дня мы создали свои каналы и начали писать! Вот этого пенделя мне и не хватало чтобы начать. 😁 а ещё Влад системный аналитик и ведёт свой канал.💪

Когда несколько дней пытался заставить работать интеграцию со внешним сервисом со сложным API и с неочевидной документацией. Потом оно заработало, но высосало все силы...

Люди делятся на два типа:
1️⃣ Больше никогда не буду брать задачи связанные с этим сервисом! Нахрен надо! Столько возьни! 🤬
2️⃣ Теперь я знаю как это работает, сделаю заметки, напишу комменты в коде и в следующий раз быстро решу задачу, связаную с этим сервисом! 🙌

К какому типу относитесь вы?
1 - 🤬
2 - 👏

Clean code приёмы в Airflow

📝 Делаем код с определением дага красивее 😉

Поделюсь с вами простой фишкой для передачи результата из одной таски в другую в рамках дага airflow, которая делает код чище.

У инстанса task в Airflow есть свойство output, которое содержит вывод функции вызванной в рамках задачи. Наример, если это PythonOperator, то там будет то, что вы вернули в через return. Посмотрим пример 👀

Обычно некрасивый код выглядит примерно так

def generate_data():
    return {"event": "order closed", "project": "superpuper", "value": 2}


def process_data(**kwargs):
    # Access the previous task's result using ti
    ti = kwargs['ti']
    task_output = ti.xcom_pull(task_ids='generate_data_task')  # Access the result
    processed_value = task_output["value"] * 2
    print(f"Processed value: {processed_value}")

# Define the DAG
with DAG(
    dag_id='example_ti_result_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
) as dag:

    # Task 1: Generate data
    generate_data_task = PythonOperator(
        task_id='generate_data_task',
        python_callable=generate_data,
    )

    # Task 2: Process data
    process_data_task = PythonOperator(
        task_id='process_data_task',
        python_callable=process_data,
    )

    # Define task dependencies
    generate_data_task >> process_data_task

Обратите внимание на явную зависимость функции process_data от оркестрации: функция вынуждена знать о деталях реализации контекста airflow, о том что существует xcom, что существует инстанс task под названием generate_data_task. Жесть! 😭

Теперь красивый код

def generate_data():
    return {"event": "order closed", "project": "superpuper", "value": 2}

def process_data(event):
    processed_value = event["value"] * 2
    print(f"Processed value: {processed_value}")

with DAG(
    dag_id='example_output_property_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
) as dag:

    generate_data_task = PythonOperator(
        task_id='generate_data_task',
        python_callable=generate_data,
    )

    process_data_task = PythonOperator(
        task_id='process_data_task',
        python_callable=process_data,
        op_kwargs={'event': generate_data_task.output},  # Using the .output property
    )

    generate_data_task >> process_data_task

Здесь видно что функция process_data стала независимой и явно принимает конкретный объект event и работает только с ним, выполняет свою задачу и ничего не знает лишнего. Детали о том, что оказывается generate_data_task возвращает именно event и что его надо передать в функцию process_data_task для дальнейшей обработки держит только оркестратор. ☝️

📝 Главный эффект такого чистого кода в том, что каждая функция становится независимой и принимает явно нужные для ее работы аргументы и ничего не знает об оркестраторе, других задачах в даге, зависимостях от других задач и последующих задачах. Каждая функция изолированно выполняет свою работу и все. Это позволяет отделить оркестратор и логику в функциях, а значит функции можно переиспользовать много раз в разных дагах или вообще без дагов и airflow! Для самой функции не имеет значения как ее вызвали и кто передал аргументы: вы руками для теста, или какой-то оркестратор, или что угодно еще. 🙌

Бонус
Когда писал этот пост попробовал спросить у ChatGPT дать пример дага с передачей результата из такси в таску, намеренно не упоминая свойство output. Невероятно, но он так и не предложил использование этого метода, даже после нескольких моих запросов типа “а как еще можно?”. 🫣 Только после того как я явно ему написал, типа “что ты знаешь о свойстве output?” он таки написал мне что да, это классный метод и изолирует все ручные операции и отметил readability, seamless, avoiding manual handling, simplifying dependencies. Я, если честно офигел, что он сам это не предложил! 😮
Так что ловите крутой прием, о котором даже chatgpt вам не подскажет! 😎

©️Если интересны и другие подобные приёмы и фишки, то ставьте сердечко и напишу ещё. ❤️

Немного творчества в преддверии Нового года.

Всех с наступающим! 🎆🥂🎁🎄

Написано в соавторстве с chatgpt. 😁 он немного помог, накидав варианты и возможные рифмы.

Clean code приёмы в Airflow. Lambda в таске

📝 Еще одна возможность определения таски в даге airflow, которая может быть полезна в некоторых случаях для небольшой подготовки данных для последующих задач.

В прошлый раз я писал об использовании свойства output.

При использовании PythonOperator в аргумент python_callable можно передать lambda-функцию.
Использование таски с лямбдой позволяет:
1️⃣Избавиться от определения отдельной функции для простой однострочной обработки
2️⃣Выполнить операцию один раз и затем переиспользовать ее результат в других задачах
3️⃣Подготовить аргумент для последующих функций явно в том виде, в котором они ожидают
4️⃣Гарантировать, что функции явно работают с нужными аргументами без лишних зависимостей

Вот пример с извлечением списка идентификаторов заказов и его использованием

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime


def generate_data():
   data = [
       {"order_id": 1, "event": "order closed", "project": "superpuper", "value": 2},
       {"order_id": 2, "event": "order closed", "project": "superpuper", "value": 25},
       {"order_id": 3, "event": "order closed", "project": "superpuper", "value": 38},
       {"order_id": 4, "event": "order closed", "project": "superpuper", "value": 4},
   ]
   return data


def get_additional_data_1(orders: list):
   # some logic to get additional data using order_id list
   # the function worked with orders list explicitly
   data = []
   return data


def get_additional_data_2(orders: list):
   # some logic to get additional data using order_id list
   # the function worked with orders list explicitly
   data = []
   return data


with DAG(
       dag_id='example_output_property_dag',
       start_date=datetime(2023, 1, 1),
       schedule_interval=None,
       catchup=False,
) as dag:

   task_generate_data = PythonOperator(
       task_id='generate_data',
       python_callable=generate_data,
   )

   task_get_order_id_list_from_data = PythonOperator(
       task_id="get_order_id_list_from_data",
       python_callable=lambda data: [row['order_id'] for row in data],  # use lambda
       op_kwargs={
           'data': task_generate_data.output, # pass argument to lambda
       },
   )
  
   # then you can use task_get_order_id_list_from_data.output to get the list of order_id

   task_get_additional_data_1 = PythonOperator(
       task_id="get_additional_data_1",
       python_callable=get_additional_data_1,
       op_kwargs={
           'orders': task_get_order_id_list_from_data.output,  # using order list
       },
   )

   task_get_additional_data_2 = PythonOperator(
       task_id="get_additional_data_2",
       python_callable=get_additional_data_2,
       op_kwargs={
           'orders': task_get_order_id_list_from_data.output,  # using order list
       },
   )
  

task_generate_data >> task_get_order_id_list_from_data >> [task_get_additional_data_1, task_get_additional_data_2]

И снова отмечу здесь признаки чистого кода:
🔸функции task_get_additional_data_2 и task_get_additional_data_1 явно работают с аргументом orders,
🔸операция извлечения списка id заказов выполняется один раз в одном месте и затем переиспользуется. 🙌

©️ Может показаться что использование лямбды таким образом очевидно, но по-моему это один из тех случаев, когда возможность на самом деле не очевидная, но когда понимаешь, что она есть, кажется что это очень просто, логично и очевидно. 😉

Извлечение внешних данных vs подготовка данных как продукта

📝 В обоих случаях есть некий пайплайн, который берет данные где-то и складывает куда-то. Но разница существенная. К техническому решению применяются требования. Инженерные требования. О них вам не скажет заказчик. Вот парочка случаев с разными требованиями к пайплайну.

✳️Данные как продукт. Наш продукт. Тут требований много. Мы отвечаем за качество и должны его гарантировать. Здесь писал про консистентность, согласованность и целостность. А в этом посте больше остановлюсь на втором случае.

✳️Данные из внешних источников.
Нужно гарантировать, что все данные будут получены и сохранены как есть. Ничего не потеряется, ничего не изменится. Эти данные как правило уже после трансформируются и превращаются в наш продукт.

Ключевая разница в разном уровне гарантии. ☝️

По внешним данным мы не можем гарантировать валидность, целостность, а только консистентность - соответствие источнику. Источник их опубликовал, а значит считает их достаточно качественными. Валидны данные или нет, целостны или нет с нашей точки зрения - разберемся потом.

Как это влияет на технические решения на практике.
🔸При трансфере внешних данных нет трансформаций! EL, если хотите, - extract and load. Как только вы вклинились с какой-то своей логикой посередине уже гораздо сложнее гарантиговать консистентность.
🔸Для внешних данных минимальны проверки при записи. Даже тип данных лучше не проверять. Если отправили текст вместо числа, так и запишем - значит так надо. 🤷‍♂
🔸Это не наш продукт и эти данные мы не публикуем! Это как бы сырье, которое мы применим или нет, чтобы создать наш продукт, за который уже будем отвечать и гарантировать качество.

🔍Если хотите порассуждать чуть глубже, то как и всегда есть одно но. Говоря красивыми словами, эти внешние данные просто существуют в другом контексте. Из этого следует, что внешними данными могут быть и данные другого департамента вашей компании, который работает в своём контексте. Данные его соответствуют некой модели того контекста. Это значит что они могут быть опубликаваны нами для анализа, но только в рамках нужного контекста и тот кто будет на них смотреть тоже должен быть внутри контекста. Простыми словами он должен знать что за сущности, свойства и записи тут есть, почему они такие и какие тут могут быть допущения. Используя эти данные аналитик принимает на себя риски связанные с некорректной интерпретацией контекста.

©️Все это можно свести к одному лаконичному правилу.
Собирая внешние данные нужно гарантировать только консистентность - записали ровно так как взяли и если что-то записали, то там где взяли равно так и было.

Clean code приёмы. Работа с датами в оркестраторе

Продолжаем работать над чистотой кода и чистотой архитектуры

📝 Трансфер данных почти всегда связан с датами или, лучше сказать, с диапазонами времени. Сегодня поговорим о работе с датами вообще.

Я всегда повторяю и здесь тоже писал, что все участники должны выполнять свою задачу и не брать слишком много ответственности. Оркестратор дергает сервисы в определенной последовательности, передавая контекст. А сервисы просто делают свою работу. Все участники знают свою роль и независимы. 👌
Вернемся к работе с диапазонами времени. Знать рассматриваемый в данный момент диапазон времени - работа оркестратора. Это часть контекста. Сервисы/операторы в свою очередь должны уметь работать с любыми запросами оркестратора - сказано вернуть данные за месяц? Вернул. За полчаса? Вернул. Они исполнители, не задающие вопросов.

Работа с датами в airflow
Очень часто работа с датами в airflow ограничивается использованием диапазона в один день с использованием переменной ds из контекста. Примерно вот так в sql

where date_column = '{{ ds }}'

Загнать себя и архитектуру в такие узкие рамки - неиначе как наступить самому себе на горло.
😭

Airflow в контексте имеет много чего интересного и в том числе разнообразные таймштампы. (Вот тут документация. Они назвали это template variables, но это не только для рендеринга темплейтов это в целом переменные контекста и им следовало назвать их context variables.)

Итак в контексте всегда есть два замечательных таймштампа: data_interval_start и data_interval_end. Это собственно начало и окончание временного интервала, определяемого заданным расписанием. Например для одного дня там будет 2025-02-09 00:00:00 и 2025-02-10 00:00:00, задав в schedule три месяца получите 2024-12-09 и 2025-02-09. Да хоть 25 минут и тогда кроме даты будет 03:10:00 и 03:35:00. Я думаю вы поняли. Смысл в том, что задать в schedule можно любое расписание и получить таймштампы в контекст. Более того это ещё и готовые объекты datetime! 🤤.

Вот так дышать гораздо легче и свободнее.👍 А теперь следите за руками. 👇
Оркестратор берет любой schedule и пробрасывает везде в виде контекста диапазоны.©️ Все сервисы написаны так, чтобы они принимали любой диапазон и делали свою работу. ©️Получаем независимый от диапазона времени пайплайн! Вся логика, вся последовательность операций, вся обработка одинакова и написана один раз.©️ Хочу запускаю каждые 8 часов, хочу каждые 2 часа, хочу каждые 3 дня. Хочу забэкфилить последние полгода одним раном? Пожалуйста! Только задай расписание. Хочу близко к реалтайм - ставлю минуты и тд. 🙌

Тут же добавляется ещё одна очень интересная возможность - динамически генерировать несколько идентичных дагов с разным расписанием. Звучит сложно, но по факту постой цикл for.😎 Это например можно использовать для дополнительных запусков пайплайна с чуть более редким расписанием чем основное, на случай чего-то непредвиденного в основных запусках. У нас примерно так и есть. Почти на каждый пайплайн есть два-три дага с расписанием: 2 часа, день, месяц. Ну и конечно бэкфил можно запускать хоть помесячно хоть поквартально.

📎 В этом посте уже не осталось места для примеров с кодом, так что покажу в одном из следующих постов, в котором покажу работу в коде с контекстом вообще и в том числе с датами. 👀

©️В качестве краткого резюме. Диапазон дат - детали. Логика пайплайна не должна меняться в зависимости от диапазона дат. Оркестратор дергает участников передавая даты, а сервис/оператор должен уметь работать с любыми датами.

Бизнесу пофиг

Когда я говорю Бизнес, я имею ввиду людей которые принимают решения, стейкхолдеров, топ менеджмент.

Бизнесу все равно какой архитектурный паттерн вы там применили, чтобы организовать данные, какие потоки настроили, какие инструменты применили и тд. Без разницы ему data vault у вас, Inmon или Kimbell. Пофиг data lake, lakehouse или warehouse.🙄

Бизнес интересуют ответы на вопросы
Им интересно:
1️⃣ Можно ответить на вопрос? Да или нет.
2️⃣ Как быстро можно ответить на вопрос? В любой момент/ через часы/ через дни.
3️⃣ Будет ли уведомление, если что-то идёт не так? (Бизнес алерты. Не путать с техническими)
4️⃣ Сколько стоят ответы на вопросы? (Стоимость инструментов, трудозатраты аналитиков и инженеров)

При этом по умолчанию подразумевается, что ответить можно на любой вопрос и что данные точные (хотя иногда надо это прояснять - писал тут)

©️Избавьте бизнес от любых технических подробностей и он вас полюбит.❤️
Да, от любых. Да, даже если они хотят залезть в технические детали, аккуратно выводите их на уровень бизнеса.