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

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️⃣ Сколько стоят ответы на вопросы? (Стоимость инструментов, трудозатраты аналитиков и инженеров)

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

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

Приоритеты не нужны! Часть 1.

Дисклеймер: Это пост — рассуждение. Здесь будет просто поток мыслей без конкретных пошаговых инструкций. Я не знаю, насколько это вам интересно, но решил попробовать.

Мне пришла странная мысль, и чем больше я о ней думаю, тем больше она мне нравится.

Мы часто говорим о приоритетах на митах. Особенно когда ресурсов мало. Мы даже собирали заказчиков вместе, чтобы они договорились о приоритетах. Батл: чья задача важнее. И в итоге мы откладывали все в сторону, чтобы сделать "важное". А то, что "неважно", откладывадось на неопределённый срок. В течение нескольких месяцев, будучи лидом, я был убеждён, что так и надо, меня так научили. Когда ко мне кто-то приходил из бизнеса, я говорил: ничего не знаю, в приоритете другое, а приоритеты ставит бизнес, то есть вы сами. Но если подумать, это супер странно. Я всегда рассматриваю работу нашей команды как сервис для бизнеса, а значит все проблемы бизнеса важны, все клиенты важны. Вот и пришла мне мысль, о которой я хочу порассуждать в этом посте.

❗️Приоритеты не нужны! Нужно выполнять ВСЕ задачи, которые прилетают команде. В разумные сроки.

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

❓Но тут возникает резонный вопрос: что такое разумные сроки? И тут, кажется, в двух словах не опишешь, но интуитивно мы это всегда как-то понимаем.
▫️Например, есть задача, которая точно терпит и может быть отложена на месяц, но занимает пять минут, и вы решаете сделать это сразу.
▫️Или бывает проблема, решение которой даст больший эффект, чем другая, но заказчик первой готов ждать дольше, чем нужно времени для решения. Тогда вы сначала берете другую задачу — с меньшим эффектом, но с большей вовлечённостью заказчика (это может быть большее давление или, наоборот, просто хорошие личные отношения. Глупо игнорировать тот факт, что мы в конце концов работаем с людьми.)
▫️Или еще бывают инфраструктурные задачи, у которых нет заказчика, но они важны с инженерной точки зрения, а точнее с точки зрения улучшения качества услуг в будущем (быстро, качественно, недорого). Сюда же я бы отнес и обучение и обмен опытом.
▫️Ну и конечно куда же без хотфиксов: пожары нужно тушить здесь и сейчас.

✔️ Выходит, что разумный срок — это некое сочетание требуемых трудозатрат, степени влияния проблемы и вовлеченности заказчика.

Смотри следующий пост ⬇️

Приоритеты не нужны! Часть 2.

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

❓Следующий резонный вопрос: что делать прямо сейчас, если у команды/ у меня сразу несколько задач?

Получается нужно попилить доступные ресурсы (время команды, ваше время) на задачи так, чтобы все задачи были выполнены в разумный срок. Возникает необходимость распределения ресурсов. И часто как раз для этого используют приоритеты, типа у этой задачи приоритет выше только её и делаем. Но тут мы говорим о том, что для нас все клиенты важны и все задачи должны быть выполнены, поэтому не подходит.
В идеале для каждой задачи определить разумный срок и необходимые трудозатраты и выделять время пропорционально: час в день, половину вторника или четверг и среду на одно, а понедельник, вторник, пятница на другое. Важно остановиться где-то посередине между двумя крайностями:
❌ много задач решаются параллельно, то есть буквально одновременно, на одном мониторе у вас одна, а на втором другая, пару строк там пару тут - нет фокуса и слишком частое переключение - будут ошибки и усталость.
❌ проблемы решаются от начала до конца последовательно - тут как раз все проблемы, описанные выше: пока делаешь одну важную, другая ждет, потом появляется ещё одна более важная и та другая ждет бесконечно. И ещё в таком подходе большой риск связанный с ошибкой в оценке сроков или в оценке влияния проблемы: решение "важной" проблемы затянулась по какой-то причине, а другая так и не была начата. Или со временем бизнес решил, что та другая все таки важнее, а вы три недели занимались первой.
✔️Баланс между этими крайностями - ключ к высокоэффективной и гибкой системе доставки результата клиентам. Особенно если удастся разбить большую проблему на мелкие части и выдавать их по мере выполнения.

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

🔍 выглядит так, что нужно расписание, а не приоритезация. Или я бы даже сказал оркестрация 😀

❗️Приоритеты не нужны! После этих рассуждений я собираюсь исключить из диалогов с бизнесом и командой слово "приоритет". Мы оказываем услуги, поэтому все запросы важны☝️

Бонус, для тех кому интересна тема + косвенное подтверждение состоятельности рассуждений. Инжиниринг он везде инжиниринг. Многие принципы и лучшие практики наследуются из разных областей, и бережливое производство Toyota — пожалуй, один из самых известных подходов, который вышел далеко за рамки применения только на сборочных линиях.
Рассуждения в этом посте хорошо укладываются в применение метода "хейдзунка" бережливого производства Тойота. Это метод выравнивания потока, чтобы заказчики получали разные изделия равномерно небольшими партиями (в идеале единицами), вместо того, чтобы ждать, пока крупнопартийное производство дойдёт до изготовления нужного изделия - очень похоже на проблему с приоритетами. А проблема быстрой переналадки очень похожа на проблему фокуса на задаче. Тот кто умеет быстро переключаться, более гибок и способен построить более равномерный поток.

Я не знаю насколько такой формат кому-либо интересен, поэтому ставьте ❤️, если было интересно и вы задумались вместе со мной.

Clean code в Airflow. Контекст

Продолжаем делать код в Airflow красивее и чище 😉

Работа с контекстом - ещё один очень простой способ снизить зависимость логики от оркестрации и на шаг приблизиться к Clean code.

Airflow создаёт контекст при каждом запуске дага. Этот контекст содержит ряд аргументов, которые могут передаваться оператору. Это по сути key-value пары или в случае python - dict. Оператор может использовать их для рендеринга темплейтов или в качестве аргументов функций. Вот как раз про передачу контекста как аргументов в функции я и хочу написать в этом посте.

📝 Если в двух словах — передавай контекст, принимай аргументы☝️. Код становится более читаемым и главное чистым, когда вместо kwargs принимаются конкретные аргументы.

Вот пример при работе с датами (пост про даты) и не только.

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


def extract_data(
        data_interval_start: datetime,  # забираем из контекста только то, что нужно
        data_interval_end: datetime,  # типизация поможет IDE найти проблемы в коде
        project: str = None,  # опциональный параметр
        **kwargs  # поможет игнорировать лишние аргументы из контекста 
                  # особенно полезно, если функция используется в нескольких дагах
)-> list[dict]:
   data = []
   return data


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

   task_extract_data = PythonOperator(
       task_id='extract_data',
       python_callable=extract_data, # контекст передается всегда, то есть никакие действия не нужны
   )

   task_extract_data2 = PythonOperator(
       task_id='extract_data2',
       python_callable=extract_data,
       op_kwargs={
           'project': 'the_best_of_the_best',  # можно добавить любые аргументы
       },
   )

Как видите не нужно как-либо явно передавать аргументы, которые уже в контексте. Они только и ждут, чтобы их красиво принимали🤤. Опять же отмечу, что, пожалуй, самый важный эффект здесь — снижение зависимости функции от оркестратора: функция явно принимает всё необходимое для работы. Как бонус — это удобно при работе в IDE: теперь IDE может анализировать код и подсвечивать проблемы, особенно если вы укажете типы.😎

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

Вот ссылки на другие посты о Clean Code в Airflow
1) передача результата между тасками и свойство output
2) lambda в таске
3) про работу с датами

Про что одна строка?

📝 Пожалуй самый первый, главный вопрос при работе с любым набором данных (датасетом, дата мартом (витриной), таблицей) - про что одна строка?

На самом деле это вопрос моделирования, но даже обычный разовый отчёт имеет свою модель.☝️

При этом неважно, говорим мы про подготовку данных или анализ — без понимания единицы (minimum grain) невозможно работать. Поэтому это первое, что нужно для себя прояснить, даже если заказчик не может ответить на вопрос прямо. 💡

Язык данных = бизнес язык

📝 Если бизнес использует какие-то понятия или свойства, добавьте их в модель явно.

Нужно внимательно слушать как разговаривает бизнес и сделать так, чтобы данные могли явно отвечать на их вопросы. Ответы на вопросы бизнеса должны быть лёгкими и непринужденными.
Кстати из этого вытекают требования к модели данных. Я бы даже сказал, что в этом заключается искусство моделирования.

Чем лучше продумана модель, тем проще селект запрос к данным для ответа на вопрос.❤️ В идеале в селекте должны быть только джойны, агрегации и фильтры. И всё это просто, без подзапросов, кейсов и прочих трансформаций в селекте.

Пример
На мите вы слышите от бизнеса такую фразу:

лиды, которые позвонили нам по общую линию, могут иметь разное намерение (купить или получить поддержу), которое еще должно быть подтверждено.

В одной фразе заключена ценнейшая информация о бизнес модели! 👀 Есть сущность лид. Лид имеет свойство источник, свойство намерение, и свойство-флаг подтверждено ли намерение. Вот что должно быть в данных! Определения этих понятий, конечно, следует уточнить и в идеале записать в документ. Этот документ и есть бизнес модель данных. Не путать с табличками в базе данных приложения.
Как правило, в базе данных приложения техническая модель: лид создан через API, звонок был на какой-то номер, плюс некоторые признаки намерения и его подтверждения, которые ещё нужно правильно интерпретировать, то есть трансформировать.

Или вот бытовой пример. Бизнес просит посчитать бананы, а в базе приложения у вас только желтые объекты продолговатой формы.😁 Вот и назовите их в модели явно бананами и убедитесь, что точно отделили их от кукурузы.🙌

©️Задача инженера перевести данные с технического языка на бизнес язык.

Такие разные select запросы & трансформации

📝В прошлом посте я написал такую фразу:

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

В этом посте я разовью эту мысль и порассуждаю о трансформациях в обращении к данным. 🤔

Select запросы к данным бывают разные: с трансформацией и без.
Для ответа на вопросы (отчеты) трансформаций быть не должно!

Вот признаки трансформации в селекте:
🔵 Применение логики/правил, видно по использованию if, case, coalesce и других условных операторов.
🔵 GROUP BY используется для дедупликации, а не для агрегации.
🔵 WHERE используется для исключения некорректных данных, а не для выборки.
🔵 Есть подзапросы. Любые. Там, где JOIN, или там, где колонки, или через CTE (WITH) — неважно.
🔵 Изменение значений. Например, назначение имен идентификаторам статусов или переименование текстовых значений.
🔵 Расчёты и формулы.
🔵 Использование оператора LIKE.

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

Тут вы можете справедливо заметить, что бывают новые запросы, к которым данные ещё не готовы, или проверки гипотез. И то, и другое, конечно, потребует трансформаций в запросах, но это по сути — ресерч, проверка и подготовка нужной логики, и затем принятый вариант всё равно должен быть автоматизирован и превращён в витрину: это гарантирует, что трансформация всегда выполняется в одном месте и один раз, и любые изменения наследуются всеми зависимыми моделями.🙌 Кроме того логика трансформации может быть вынесена за пределы sql и выполнена на python, в spark и тд. А витрина может быть опубликована через API. Суть от этого не меняется. ☝️

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

💡 Если для регулярных отчетов требуется трансформация - делайте витрину. Если есть более одной витрины с похожей логикой трансформации - пересмотрите модель данных.

©️не все select запросы одинаковы по смыслу. Для предсказуемых ответов на вопросы Select должны быть без трансформаций 👌

Correlation vs causation

Это труба погнулась, потому что села чайка? Или чайка села именно на место сгиба уже после того, как труба была согнута?

В этом случае ответ, конечно, очевиден. Но эта картинка отлично подчёркивает саму идею:

📝 то, что выглядит на первый
взгляд как причинно-следственная связь, может оказаться простым совпадением или следствием какой-то иной, настоящей причины.

Об этом стоит помнить при решении проблем. Особенно при траблшутинге (это когда что-то не работает, и срочно нужно сделать так, чтобы работало😁).

Например: это вот этот тяжёлый, кривой неоптимальный запрос повесил базу данных, или что-то нагрузило базу данных, а запрос просто оказался не в том месте и не в то время — прям как чайка на картинке.

Такие картинки кстати здорово всплывают в памяти в нужный момент и помогают посмотреть на ситуацию под другим углом. Наверное, поэтому они и называются мемами (на самом деле все сложнее и интереснее, спросите у gpt откуда слово meme). Так и я год назад увидел эту чайку, и теперь каждый раз её вспоминаю, когда нужно разобраться, что пошло не так - correlation vs causation. 🤔 Надеюсь и вам пригодится.

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

Meta data или данные о данных

✏️ Meta data - это данные о данных. То есть это информация именно о самой записи, о строке: откуда она взялась, кто записал, когда создана запись, когда изменена и многое другое.

Это важно знать и записывать, чтобы иметь больше контроля над данными, чтобы знать больше деталей о конкретно этой записи (строке) в этой таблице. Иметь её "паспорт". 💪

На практике это просто дополнительные колонки в той же таблице. Ну или если у вас json, то дополнительный meta объект внутри. При этом их стоит явно отделить от данных, например используя префикс. Мы вот используем meta_.

Еще раз подчеркну, что важно не путать мета данные и данные!
Вот несколько примеров, объясняющих суть различия:

🔵 заказ создан 1 мая 21:00, а запись в DWH о нем создана 1 мая 22:35 (потому что у вас периодический батч процесс), created_at заказа будет отличаться от meta_created_at. А если потом уже внутри DWH на основе таблицы заказов будет обновлен дата март, то там опять же будет свое meta значение, например 22:45. Дата создания заказа не меняется, а вот дата создания записи о заказе в той или иной таблице будет своя.

🔵 Еще более наглядный пример при бэкфиле. Заказ создан в приложении 1 мая 2025 года. По какой-то причине данные о нем собраны вчера - 2 июня 2025. Важно знать, что эта запись была создана только вчера, и что отчеты построенные, например, 1 июня еще ничего о нем не знали! ☝️

🔵и ещё пример. вы сохраняете run id - идентификатор процесса, в рамках которого создана/изменены строка. Это поможет, например, при траблшутинге, так как вы точно знаете какие данные затронул тот самый ран.

В метаданные можно действительно засунуть много всего, и то, что я перечислил выше, — лишь малая часть того, что можно придумать. Все зависит или от проблем (в том числе потенциальных), которые вы хотите решить с их помощью, или от сложности потоков данных, которые ведут к таблице, особенно если там несколько пайплайнов. 😵‍💫
В каких-то таблицах мы накидываем чуть больше мета данных, в каких-то меньше.

Я даже думаю, что если у вас еще нет мета колонок в данных и вы считаете, что они вам не нужны, то просто вы еще не созрели 😁 Вы почуствуете этот момент, когда они вам понадобятся. Даже если вы просто собираете отчёт, стоит как минимум знать дату последнего обновления строки, которую вы взяли в отчёт.🫣

Минимальный набор по-моему должен включать две колонки:
©️meta_created_at
©️meta_updated_at
Я бы прям рекомендовал вставлять их во все таблицы. Для зависимых (downstream) процессов необходимо знать именно изменённые записи и meta_updated_at будет как раз кстати.👍

Ниже список колонок, которые мы используем в разных местах в зависимости от того, что мы хотим знать о записях табличке. Я сгруппировал их по "участнику" потока, через "руки" которого прошли данные, и он добавил свои мета "следы".

📎 Если есть записи из разных источников:
🔵meta_source
🔵meta_source_object_type
🔵meta_source_object_id

📎Об оркестраторе:
🔵meta_orchestrator_run_id
🔵meta_orchestrator_run_timestamp

📎 Хранилище:
🔵meta_created_at
🔵meta_updated_at

Плюс если храним историю изменений:
🔵meta_is_current
🔵meta_valid_from
🔵meta_valid_to
🔵meta_version_hash
🔵meta_version_number
🔵meta_version

Я напишу на эту тему ещё пару постов, например чуть подробнее напишу про meta_updated_at 🙌 так что продолжение следует... 😊

Meta данные - meta_updated_at. Часть 1

В прошлом посте писал о мета данных в целом, а в этом двойном посте остановлюсь на реализации одного из самых важных мета свойств - таймштамп последнего изменения. ☝️

Будет длинный двойной пост.
Конечно, я мог бы просто написать как мы делаем, но так неинтересно, по крайней мере мне самому 😁 я напишу почему именно так. Спойлер: во второй части будет код функциии на postgresql 👍

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

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

Для начала нужно решить кто ответственен за эту свойство. У нас есть три (как минимум) участника, через "руки" которых проходят данные и которые могут добавить свой мета след:
🔵источник
🔵исполнитель (оркестратор)
🔵хранилище.

Порассуждаем: чтобы сказать изменится ли что-то, нужно знать в каком состоянии это что-то сейчас.

Источник ничего не знает о том, в каком состоянии сейчас запись в месте назначения, более того он вообще не знает о существовании этого самого места назначения. Есть соблазн в момент извлечения данных добавить прям в select что-то вроде now или current timestamp, но это будет время извлечения, а не каждое извлечение = изменение!

Примерно по той же причине этого не может знать и исполнитель (оркестратор). Он может только знать о времени обработки, что опять же не тоже самое, что время изменения.

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

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

Вторая часть поста⬇️