Where is data, Lebowski

🏖Избавляемся от графика опусков
.

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

.

Давай скажем, что порядок предоставления отпусков у нас называется: "Возьми сам!"

.
Конечно, это классная идея: объявить людям, что они вправе распоряжаться собственной жизнью и могут решать, когда им работать, а когда брать паузу. Логично возникают в кошмара:
1️⃣ Все сотрудники взяли отпуск одновременно и работать некому
2️⃣ Никто не ходит в отпуск и все перерабатывают

Как предлагается преодолеть кошмары:
2️⃣ Если глава компании отдыхает всего пару недель, разумеется, сотрудники будут чувствовать, что свободный график на деле не предполагает особой свободы. В отсутствии единой политики большинство людей смотрят на окружающих и пытаются вычислить "мягкие границы" дозволенного.
Поэтому решение: при желании внедрить в компании свободный график отпусков, начните с себя: Руководители должны регулярно отдыхать сами!
.
1️⃣ .. для успешного ослабления контроля без ущерба для дела необходимо предпринять еще один щаг. Когда отменяется протокол или регламент, то сотрудники не знают, как вести себя. Некоторые впадают в ступор, пока им начальник не объяснит, что конкретно нужно делать, а чего нет. Других опьяняет чувство вседозволенности и они могут повести себя непредстказуемым образом - например, взять отпуск в самое тяжелое для команды время. Руководителю необходимо как можно чаще проговаривать условия, на которых его подчиненные смогли бы поддерживать оптимальное соотношение своей работы и досуга

Рауиональное зерно здесь есть: "глубина" отдыха (скорее речь про его качество, а не кол-во) должна соответствовать кол-ву стресса от рабочего периода.

Что думаете, можно такое построить (безотносительно законодательных мер) или гоу все заполнять просроченные графики отпусков на 2025 год?

#netflix #norules

Серия правил продолжется...

Стоит сказать, что правила выбираю те, что показались наиболее интересными и без всех подробностей, настоятельно рекомендую прочесть книгу (хотя бы ознакомиться с примерами 😉)

#netflix #norules

🗣 Говори то, что думаешь (с благими намерениями)
.
Многие из нас бывали в ситуациях, когда действия/идеи коллег вызывают желание высказать критику, но скорее всего по следующим причинам выбираем тактичное молчание:
- Вы боитесь, что вас не поддержат
- Вы не хотите, чтобы за вами закрепилась репутация скандалиста
- Вы не хотите неприятных разговоров
- Вы не хотите расстроить или обидеть коллег
- ...

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

.
Кажется, что в такой дозволенности можно наговорить, что угодно, испортить отношения с коллегами и тд. Доказательная база такова: в 2014 году агенство Zenger Folkman опросило ~1000 человоек. Примерно четверть скаазали, что критика гораздо эффективнее помогает исправить недочеты и добиться успеха, несмотря на то, что позвала дарит блаженство. Большинство не отметило никакого влияния на позвалы на качество их работы. Что еще интересного было в том опросе:
- 57% респондентов предпочли бы получать конструктвную критику, а не положительные отзывы
- 72% сочли, что если бы им чаще указывали на недочеты, качество их работы улучшилось бы
- 92% согласились с высказываением: "Грамотная критика помогает"
.
С этой цели компания Netflix постоянно работает со своими сотрудниками в части как правильно высказывать и принимать критику, на основе этих работ сложилась некая система правил.
.
Четыре правила откровенности
1️⃣ Для критика:
- Стремись помочь: критиковать нужно с благими намерениями, срывать злость, унижать собеседника или преследовать собственные интересы недопустимо. Нужно объяснить, каким образом перемена в поведении поможет адресату критики или всей команде
- Предлагай конкретные меры: необходимо указать что именно должен изменить адресат критики

2️⃣ Для адресата:
- Будь благодарен: получив критическое замечание, нужно справиться с естественной реакцией присихики (напряжение, инстинктивная зашита, подготовка к ответному удару) и спросить себя: "Как я могу проявить благодарность за отзыв о моей работе" - выслушать, тщательно обдумать, воздержаться от протестов и возражений
- Прими или отклони: Замечаний или критики может быть множество, вы не обязаны принимать их все как руководство к действию. Вежливо и искренно поблагодарите критика, но и вы, и он должны понимать: следовать или нет - выше и только ваше решение
.

Ну что, как вам правило, есть тут здравое зерно?
#netflix #norules

На старт, внимание, марш -> Никаких правил от Netflix🔥

Сначала правило -> краткое описание -> думаем, обсуждаем

#netflix #norules

​​🎅 Разибраем подарки от деда Мороза
.
Одним из приятных подарков стала книга Никаких правил.Уникальная культура Netflix📚. Давно было желание её прочесть или схожую с ней про историю Netflix Инсайдерская история - учиться надо у лучших или на своих ошибах.
.
Книга рассказывает о корпоративных правилах, затрагивая историю становления и развития компании. Показывает как менялись или появлялись правила, объясняет их смысл, всё это подается с интересными историями, показывающими правила в действии или, наоборот, куда приводит отсутствие таковых.
.
Правила довольно интересные: от тех, которые сразу вызывают отторжение до принимаемых абсолютно, но чем больше ты погружаешься в них, то тем сильнее ты проникаешься отторженными и начинаешь их принимать - довольно удивительное наблюдение.
.
Давайте вместе погружаться в мир корпоративных правил Netflix 🙌

#netflix #book

🔔Там на невиданных дорожках...
.
Первое доброе дело в новом году - расчищенная дорожка 🏖
.
Пусть в новом году у вас будут прямые дорожки и пути к вашим мечтам и достижениям🙌
Пусть на расчищенных дорожках вас встретят те двери, которые вы хотели открыть или те, в которые хотели заглянуть👀
.
(как и в конце этой дорожки есть баня) Пусть новый год позволит вам обновиться до желаемых высот или очистится от суеты и увидеть эти высоты, будьте счастливы🔥☕️

#ny #newyear #2025

Последний рабочий день уходящего года прошёл под девизом песни БГ Этот поезд в огне 🚂
.
А как прошёл ваш день, перетащили всё задачки в Done😉

#work #meme #ny

Пост Новогоднего настроения 😎
.
Нравиться ли вам, когда ПО меняет иконки или добавляет новогодних эффектов.
Примеры на скрине:
- dbeaver
- confluence
- polaris (управление умной бытовой техникой)
.
У меня вызывает это детский восторг 😛 + 100500 к силе 💪, приятно что пользовательский опыт становиться положительным.
А как вам такие новогодние изменения? Если есть еще примеры кидайте в комменты...

#ny #2025

​​

from models import ExtractorResource


class MockExtractor:
    def init(self, integration_metadata: dict):
        self.integration_metadata = integration_metadata

    def get_resources(self):
        for idx in range(5):
            yield ExtractorResource(path=f'mock_s3_file_{idx}.csv')

ps: можно заматить, экстрактор реализован как генератор (yield) - в конкретном случае не так важно, тк экстрактор не отдает сами данные, а только ссылки (путь на S3 или url и тд)

Остальные классы, реализуем самостоятельно или заглядываем в репу

Каждый из классов имеет единственный аргумент = содержимое yml конфига и ничего более (но это не точно 😉), то есть все необходимые параметры должны быть описаны
в теле самого yml в нужной секции, например, у меня получилось так для экстрактора:

    tasks:
      extractor:
        MockExtractor:
          src_s3_conection_id: reddit_s3_connection_id
          src_s3_bucket: raw-public
          src_s3_prefix_template: reddit/{dm_date}
          src_s3_partition_fmt: '%Y-%m-%d'

Осталось встроить наши классы в даг, используем TaskFlow API, пример для экстратора:

@task
def _extractor(extractor: t.Callable) -> t.List[str]:
    extractor_obj = extractor(
        intergation_metadata=intergation_metadata
    )

    return [resource.dict for resource in extractor_obj.get_resources()]

# извлекаем общую структуру тасок
tasks_meta = intergation_metadata.get("tasks", {})

# извлекаем extractor
extractor_name, extractor_params = list(tasks_meta.get("extractor").items())[0]

if not extractor_params:
    extractor_params = {}

logging.info([extractor_name, extractor_params])
extractor = globals()[extractor_name]

# возвращаем список объектов
ext_resources = extractor.override(task_id=f"extractor{extractor_name}")(
    extractor=extractor)

Комментарии:
- декораторная таска принимает только саму фукнцию или класс в нашем случае
- имя экстратора получаем из ямла
- все экстраторы импортируются from extractors import *
- из словаря globals() получаем объект нужного экстрактора по имени из ямла
- если параметров экстратора нет, то подставляют пустой словарь
- resource.dict - нужно, тк XComm не знает как сериализовать нашу модель ресурса, поэтому воспользуемся атрибутом dict. Соответственно внутри таски transform_and_save обратно создадим ресурс

Логика для трансформера и saver сохраняется, добавляется только обработка ситуации,
когда трансформер сохраняет сам объекты и отдает только пути (пустой TransformerResource.content). Как показала практика: такое нередко
встречается.

Остальное обдумываем сами или заглядываем в репу.

И после загрузки в AirFlow получаем красоту в UI - repo

#walle
#framework
#automate

WALLE - Выстраиваем структуру

В прошлом посте собрали фундамент:
- сформировали yml
- научились его читать
- научились создавать даг из yml метадаты

Пока в даге только пустые таски (start\end), сейчас будем исправлять это. Во вступительном посте
очертили архитектуру дага, она повторяет процессы E(xtract)T(ransform)L(Save).
Таски transform, save предлагаю объеднить, тк transformer отдает какие-то данные, а saver их сохраняет, то есть таски обмениваются данные, а делать это через
XComm неблагодарное дело, поэтому наши даги будут состоять из двух тасок (минимум):
- extractor
- transformer_and_saver


Исходные данные (например, выгрузка API или файл на S3) в единственном числе (в смысле, что данные ASIS могут быть только одни) поэтому
extractor всегда 1, но может возвращать несколько объектов (например, N путей до файлов)

А вот обработать данные мы уже можем несколькими способами, поэтому transformers может быть несколько, а saver всегда идет в комплекте к transformer.

Для описания тасок выделим секцию tasks в нашем yml:

version: 2
models:
  - name: mock # имя интеграции, оно же dag_id
    description: Топ реддитов за последний час # описание интеграции, оно же dag_description
    dag:
#      dag_id: "" # можно переопределить dag_id != name
      schedule_interval: 0 * * * *
      start_date: '2024-08-01'
#      end_date: '2024-08-31'
      catchup: False
      owner: dwh
      tags:
        - mock
        - api_integration
    tasks:
      extractor:
        MockExtractor:
      transformers:
        - MockTransformer:
      saver:
        MockSaver:
    alerting_chat_id: -987654321
    alerting_secret_name: alerting_bot_token

Тут важно проверить, что ошибок в структуре нет и yaml_reader успешно читает такой конфиг (самостоятельно).

Структуру описали, теперь разбираемся что же это за MockExtractor, MockTransformer и MockSaver🤔. А этих товарищей нужно реализовать: то есть
это некие Python-классы, которые реализуют некоторый базовый интерфейс. На текущий момент мы знаем, что extractor что-то передает transformer, этот
в свою очередь передаёт уже данные в saver. Условимся называть то чем обмениваются классы ресурсом (Resource). Итого у нас будет 3 ресурса:
- ExtractorResource
- TransformerResource
- SaverResource

Ресурс - это объект Python, будем использовать датаклассы (dataclasses), но до них доберемся чуть позже. А сейчас про функции каждого объекта:
- Extractor (выгружает из API и складывает на S3 (это я называю RAW-слоем = данные ASIS) и отдает далее пути до файлов
или ищет наличие файлов на S3 или FTP и возвращает пути к нужным файлам)
- Transformer (читаем данные по полученным путям, перекодирует согласно нашей логике и отдаёт saver-у набор байтов для сохранения)
- Saver (просто сохраняет байтики на S3 в нужном нам формате\партицировании - это я называю ODS-слой)

Интерфейс у нас будет единообразный, поэтому опишем какие методы должны быть реализованы у каждого класса:
- Extractor - должен иметь метод get_resources - возвращает генератор объетов ExtractResource
- Transformer - должен иметь метод transform, который принимает ExtractResource, возвращает объект TransformResource
- Saver - имеет 1 метод save, который принимает TransformResource и возвращает SaveResource.

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

Расчехляем Pycharm и кодируем, ресурсы:

@dataclasses.dataclass
class ExtractorResource:
    path: str


@dataclasses.dataclass
class TransformerResource:
    path: str
    content: io.BytesIO


@dataclasses.dataclass
class SaverResource:
    path: str

Для проверки идеи и работоспособности дага создадим Mock классы, для примера MockExtractor:

​​МАТЕМАРКЕТИНГ - 2024

Запоздалый пост о конфе матемаркетинг-2024. На фото наша банда на стенде

Постояли на стенде Побывали на конференции, впечатления:
- масштабно
- с шиком\блеском
- с Себрантом
- шумно

Инженерная секция пока представлена слабо, согласно программе все рассказывали о том как
построить DataMesh. Не стало исключением и выступление нашего Head of DWH, рассказал:
- о том какие проблемы были и как решали
- как делили апельсин DWH
- где чья ответственность

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

Из интересных докладов можно отметить Рому Бунина о главном качестве аналитика. Это было не просто интересно, это было
визуально приятно, Рома в своем стиле с графиками, динамическим оформлением презы и даже вставками видео рассказал о чем не стоит
забывать аналитика (кажется, не только им):
- харды не главное
- чем выше ваш уровень (middle -> senior) тем важнее для вас софты

Посмотреть на фото Ромы тут

Кажется, это старо как мир: учитесь общаться, договариваться, задавать вопросы, говорить нет,
критически мыслить и ....


#matemarketing

​​#post

👀 Data Overview
.
Поговорим о том какие инструменты есть в Clickhouse для обзора данных. Кейсы, возникающие в работе:
- на s3 сложили файлы, какая структура, какие типы данных
- хотите выгружать данные с бека (например, Postgres) и хочется узнать в какие типы превратить атрибуты или просто узнать о структуре таблицы (а например, локального доступа нет)
- узнать структуру существующей таблицы
.
Да, в Clickhouse есть аналог information_schema - это база system, таблиц там много, но нам интересны: tables, columns - тут всё как в PostgeSQL, разобраться довольно просто.
А можно такие же, только с перламутровыми пуговицами...
.
Конечно, их есть у меня:
1. DESCRIBE - команду можно использовать не только с существующими таблицами, но и с табличными функциями:

-- local table
DESCRIBE cdc.openweathermap_raw;

name            |type         |
----------------+-------------+
record_timestamp|DateTime64(6)|
record_value    |String       |

-- files on s3
describe s3('https://storage.yandexcloud.net/public-bucket-6/reddit/2024-10-26/top_subreddits_2024_10_26_03_00_00.parquet', '', '', 'Parquet');

name                   |type   |
-----------------------+-------+
id                     |String |
title                  |String |
score                  |Int64  |
num_comments           |Int64  |
author                 |String |
created_utc            |String |
url                    |String |
upvote_ratio           |Float64|
over_18                |UInt8  |
edited                 |UInt8  |
spoiler                |UInt8  |
stickied               |UInt8  |
subreddit_name_prefixed|String |

DESCRIBE postgresql('192.168.55.94:5432', 'tmp', 'superstore_v1', 'shpz', '12345', 'stage');

name          |type                     |
--------------+-------------------------+
order_id      |Nullable(String)         |
order_date    |Nullable(Date)           |
ship_date     |Nullable(Date)           |
ship_mode     |Nullable(String)         |
customer_id   |Nullable(String)         |
custsomer_name|Nullable(String)         |
segment       |Nullable(String)         |
country       |Nullable(String)         |
city          |Nullable(String)         |
state         |Nullable(String)         |
postal_code   |Nullable(Int64)          |
region        |Nullable(String)         |
product_id    |Nullable(String)         |
category      |Nullable(String)         |
sub_category  |Nullable(String)         |
product_name  |Nullable(String)         |
sales         |Nullable(Decimal(38, 19))|
quantity      |Nullable(Int64)          |
discount      |Nullable(Decimal(38, 19))|
profit        |Nullable(Decimal(38, 19))|

... and so one. В одном из рабочих кейсов перешли от чтения паркет-файла, для формирования data_schema, к выводу DESCRIBE - 👌
.
2. SHOW CREATE - покажет вам DDL как он есть

SHOW CREATE cdc.openweathermap;

.
3. Изучить метаданные паркетов поможет указание ParquetMetadata вместо формата файла

SELECT * FROM s3('https://storage.yandexcloud.net/public-bucket-6/reddit/2024-10-26/top_subreddits_2024_10_26_03_00_00.parquet', '', '', 'ParquetMetadata')
FORMAT Vertical;
.
num_columns:             13
num_rows:                32
num_row_groups:          1
format_version:          2.6
metadata_size:           2513
total_uncompressed_size: 6182
total_compressed_size:   4068
columns:                 [('id','id',1,0,'BYTE_ARRAY','String','GZIP',435,254,'41.61%',['PLAIN','RLE','RLE_DICTIONARY']),('title','title',1,0,'BYTE_ARRAY','String','GZIP',1765,1191,'32.52%',['PLAIN','RLE','RLE_DICTIONARY']),...]
row_groups:              [(13,32,6182,4068,[('id','id',254,435,true,(32,0,NULL,'1gcanab','1gcbktk')),...]

.
ps: clickhouse не перестаёт меня удивлять, есть ощущение, что при правильном рецепте это пушка-бомба 💣

#clickhouse #lifehacks

​​Превратности панд 🐼
.
Кейс из серии явное лучше неявного. Многие сталкиваются с задачей конвертации строковой даты в собственно дату.
.
Рассмотрим пример в пандах для файлика (в нем ничего кроме даты нет):

import pandas as pd
print(pd.version)

df = pd.read_csv("sample.csv")
df["entry_date"] = pd.to_datetime(df["entry_date"])

.
И тут нас ждут тонкости, связанные с версией pandas:
- если вы используете pandas < 2.0, то вероятнее всего код выполнится без ошибок, но ошибки собирать вы будите дальше и сколько времени уйдет на поиска: часы или дни 🤷‍♂️
.
А что вообще происходит: по дефлоту пандас пытается угадать определить формат даты и преобразовать в дату. Но как выяснилось до версии 2.0 пандас делает это специфически. Открываем наш файл и видим что даты записаны как:

12/01/2018 08:26

Пока сложно: это 1 декабря или 12 января??? Поищем другие цифры и находим:

13/12/2018 09:02

Ага, значит наш формат - %d/%m/%Y %H:%M - супер. А теперь вишенка на торте, смотрим как преобразовал пандас наши строки:

12/01/2018 08:26 -> 2018-12-01 08:26:00
13/12/2018 09:02 -> 2018-12-13 09:02:00

12 января стало 1 декабря, а 13 декабря осталось 13 декабря 🤦‍♂️
.
Версии пандас >= 2.0 кидают ошибку:

ValueError: time data "13/12/2018 09:02" doesn't match format "%m/%d/%Y %H:%M", at position 881

И сразу предлагают воспользоваться аргументом format, в котором требуется указать пандасу с каким форматом даты он имеет дело.
☝️Будьте бдительны при использовании различных инструментов и старайтесь явно прописывать все настройки, не полагаясь на "умноту" этих средств - рано или поздно стрельнет в ногу.

ps: Clickhouse справился с задачей из коробки, только 🤫

SELECT
  entry_date,
  parseDateTime64BestEffort(entry_date) AS entry_date_dt
FROM
  s3('https://storage.yandexcloud.net/public-bucket-6/sandbox/sample.csv',
  'CSVWithNames')
WHERE
  entry_date IN ('12/01/2018 08:26', '13/12/2018 09:02');

entry_date      |entry_date_dt      |
----------------+-------------------+
12/01/2018 08:26|2018-01-12 05:26:00|
13/12/2018 09:02|2018-12-13 06:02:00|

#pandas #datetime

​​🧯 По следам бременских рабочих кейсов или ещё раз про ClickHouse
.
Помните решали вопрос с курсорами, чтобы никого не упасть. Упёрлись в другой нюанс, таска выполняется настолько долго, что heartbeat 💓не может прилететь, airflow со спокойной душой шлёт вас fail_alert.
.
Тут было два варианта:
- сменить executor (был кубер)
- изменить интеграцию
.
Решено было идти вторым путём. Вариантов, если cdc интеграция невозможна, кот наплакал: к именованным курсорам уже перешли ( загрузка была стабильной, но долгой, ~2часов по самой большой сущности, да, сущность прибавила в весе х2, наречем её L).
.
Хотелки инженера:
- перейти к хранению снепшотов на s3 и соответственно загрузки из s3
- dbt
- единообразие
.
Из этого уже имеется: рабочий фреймворк для запуска dbt моделей -> единообразие, осталось складировать снепшоты на s3. Был стандартный путь: читаем табличку батчами, сохраняем в файлы на s3, но хотелось чего-то другого и простого (скажем честно, хотелось экспериментов 😎).
Так у нас же ClickHouse, подумал я, и почему бы не использовать его, как compute🧐
Входные данные:
- бек postgres
- табличная функция postgresql
- табличная функция s3 (можно не только читать но и писать)
.
Получается примерно так:

INSERT INTO s3('.../bucket/entity/date=2024-10-20/data.parquet')
SELECT * FROM postgresql(host, db, table, ...)

.
Очень просто и со вкусом, а что самое интересное работает это как молния ⚡ Для сущности L время отработки запроса составляет ~15минут 🔥
.
Далее дело за малым: сделать вьюху над файлом, модельку dbt... 👉
.
Ну чем не кайфота. Такую интеграцию нельзя назвать полноценной ( тк имеем только снепшот на текущий момент, нет делитов, апдейтов и др), но если это закрывает потребность заказчика, то почему нет.
.
Будут у вас вопросики к такому подходу?
.
#work #cases #clickhouse

Рабочие мемы:
Что и когда создано, вы говорите...

#work #meme

​​WALLE - Готовим yml

Во вступительном посте оговорили, что фундамент фреймворка - yml конфиг.
Надо научиться читать его и формировать некую стркутуру для управления фреймворком.

Реализуем базу:
- все конфиги храним в директории metadata
- читаем директорию metadata
- данные из yml храним в переменной, например: integration_meta
- integration_meta подается на вход функции создания дага create_dag(integration_meta)

Какие минимальные данные необходимы для формирования дага:
- dag_id
- расписание (но это не точно)
- дата старта\окончания (мы же хотим управлять процессом)
- catchup
- куда алертить (помним, что хотим единый алертинг: в нашем случае будет телеграмм, поэтому нужен channel_id + токен)
- общая информация (owner\tags\description)

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

version: 2
models:
  - name: reddit # имя интеграции, оно же dag_id
    description: Топ реддитов за последний час # описание интеграции, оно же dag_description
    dag:
#      dag_id: "" # можно переопределить dag_id != name
      schedule_interval: 0 * * * *
      start_date: '2024-08-01'
#      end_date: '2024-08-31'
      catchup: False
      alerting_chat_id: -987654321
      alerting_secret_name: alerting_bot_token
      owner: dwh
      tags: 
        - dwh
        - reddit
        - api_integration

ps: по API будем забирать топ реддитов за последний час (данных там копейки, десятки строк, но нам не так важно кол-во)

Сохраняем наш первый yml 🥳
.

Учимся читать
.
Ничего необычного, Python библиотека pyyaml уже всё умеет.
От нас достаточно открыть файл, прочитать содержимое как текст и далее прогрузить через библиотеку pyyaml. Немного шаблонного кода:

import yaml

def read_yaml(path_to_file: str):
    with open(path_to_file, "r") as f_yaml:
        try:
            content = yaml.safe_load(f_yaml)
            return content
        except yaml.YAMLError as e:
            print(e)

Добавил обработку ошибок try\except - чуть-чуть защиты от сломанных файлов yml.

Проверяем работу реализованной функции (скрин в репе):

{'models': [{'dag': {'alerting_chat_id': -987654321,
                     'alerting_secret_name': 'alerting_bot_token',
                     'catchup': False,
                     'owner': 'dwh',
                     'schedule_interval': '0 * * * *',
                     'start_date': '2024-08-01',
                     'tags': ['dwh', 'reddit', 'api_integration']},
             'description': 'Топ реддитов за последний час',
             'name': 'reddit'}],
 'version': 2}

yaml.safe_load возвращает данные в виде python-словаря. Еще немного шаблонного кода для создания AirFlow дага и мы на финишной прямой:
- реализуем функцию create_dag, она должна возвращать объект DAG (from airflow import DAG)
- объект даг нужно положить в globals(), чтобы AirFlow смог узнать про наш даг
- пусть даг имеет следующую структуру:
- 1 таск группа
- таска start (EmptyOperator)
- таска end

Мои маленькие дата инженеры - это подлежит самостоятельной реализации 😉. Мою реализацию посмотреть тут
.
Ну и плюс, код который будет читать все конфиги из директории metadata (тут в помощь модуль из стандартной либы python os) и формировать даги. Дерзаем💪

У меня получилась такая структура папок:

- src:
  - /common:
    - func.py
  - /metadata:
    - reddit.yml
  - /yaml_reader:
    - reader.py
  - walle.py

Загружаем в AirFlow, наслаждаемся картиной (она в скринах)

На этом самая простая часть закончилась🙃

#walle #framework #automate

​​WALLE

Заботливо дадим красивое имя фреймворку Walle, почему так:
- он строит фундамент DWH ( всё отгружаем на S3)
- построение фундамента - самая грязная работа (не считая подготовки данных от RAW до STAGE слоя)
- да, это просто красиво 😍 (не называть же его YetAnotherDagGenerator или FrameWork №Х)

К предыдущему посту вопросиков накидали:

А проблемы при этом возникают, что твоя автоматизация не умеет все, что надо, её надо дописывать, 
тестить, рефакторить, иногда не работает… и это ради 10 строчек кода?

1. автоматизация не умеет всё - справедливо, но всё и не надо. Применяем правило Парето:
Автоматизируем 80% рутинной работы и оставим место нетривиальным кейсам (чтобы инженеру было где поразмять извилины 🤯)
2. рефакторить - Отвечу вопросом на вопрос: а неавтоматизированные даги рефакторить не надо? Написал один раз и забыл, нет
я такого не видел. Даже с точки зрения устаревания какого-то процесса, при автоматизации удобнее: убрал yml в архив и всё.
В общем тут посыл такой: если есть любой код - ему нужен рефактор
3. иногда не работает - справедливо, сделаю отсылку к п2: любой код может не работать =). Также вижу пользу: если не работает то, для всего - нет специфических проблем.
Очень не нравиться копаться в чужом коде, разбирая, что же там сломалось (хех, можно заметить, что если автоматизацию писал не ты, то это тоже чудой код =))
4. тестить - пффф, какие тесты. *уяк - *уяк и в продакшен
Честно, очень бы хотелось, но пока тестирование только в dev среде или на персональном стенде.


Общий посыл автоматизации такой:

Нет стремления покрыть 100% кейсов, есть желание написать код, в котором в одних и тех местах вызываются коннекшены, они имеют 
одинаковые названия, код переиспользуется (для алертинга используется 1 и только 1 функция или стандартный набор, а не 8 различных фукнций 🤦‍♂️),
таски\даги - имеют одинаковый нейминг и одинаковую структуру и тд...

#walle #framework #automate