Еще добавлю

Это будет несколько видео эксклюзивно только для тех, кто подписан на мой ТГ-канал так что приводите друзей 💡
🔤Мы рассмотрим полный путь как из ничего развернуть настоящий боевой Airflow на удаленной машине и сколько нам это будет стоить.
🔤Мы рассмотрим основные проблемы про которые все забывают (а они копятся).
🔤Мы сделаем свой репозиторий и настроим правильно CI/CD для того чтобы можно было делать комиты не ломая при этом ваш Airflow.
🔤И конечно мы сделаем первый DAG.

Никакого локального использования только боевой режим.

Потом на его базе попробуем поиграть в настоящих 🔤🔤

Всем добрейшего вторничного вечера 👏🤘

Как и обещал это первое видео из серии.
"Airflow твари и где они обедают.."

В этом видосе вы узнаете:

- Как и какой выбрать сервер под ваш могучий Airflow.
- Где арендовать и сколько деняк он вам будет стоить.
- Как зайти на сервер по SSH и не вводить пароль каждый раз.
- Где лежит докер композ для Airflow и какую версию взять.

Я сделал для Вас общий репозиторий с airflow
https://github.com/da-eos/datajungle_airflow 🔫

Там будет все что на этом видео а также там будет появляться код остальных дагов и прочие шалости 🌿

В следующих видео будем настраивать airflow.cfg и запускать докер на сервере!

Добрейшего утра четверга 😠😃

А мы с Вами продолжаем протаптывать путь к тому как делать свой Airflow на удаленном сервере.
Сегодня разбираем следующие важные шаги:

- Executor какой выбрать чем они отличаются - простыми словами.
- Добавляем в репозиторий Dockerfile + requirements.txt.
- Клонируем репозиторий на арендованный сервер.
- Делаем инициализацию базы данных для Arflow.
- Найдем где лежит airflow.cfg важнейший файл настроек вашего Airflow и добавим его в репозиторий.


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

cd <имя папки> #чтобы перейти в папку в какую то.

cd .. #чтобы выйти на уровень выше из папки.

ls -la #это просмотр содержимого в папке в которой вы находитесь, выпадает список файлов.

Так же уверен что не совсем понятно как я делаю комиты. У молодежи уже не принятно комитить через терминал все пользуют свою IDE 🙂
Но как вы успели заметить на сервере когда вы зашли по SSH нет никакой IDE 🙂 И там придется комит делать по старинке 😆
Вот как это легко делается в 4 команды.

git switch -c <ИМЯ ВЕТКИ> -- это создает новую ветку с именем что вы придумали.

git add <ИМЯ ФАЙЛА ИЛИ ПАПКИ> -- это добавляет в вашу ветки файлы что вы изменили, можно через пробел добавить несколько.

git commit -m 'Например: Добавил конфиг' -- Это сам коммит и коментарий в котором вы пишите что вы там добавили, чтобы ваши коллеги понимали что вы там сделали 🙂

## Но ваши файлы еще не в репозитории это только коммит изменений. Теперь делаем пуш в репозиторий.

git push --set-upstream origin <ИМЯ ВЕТКИ> -- имя ветки что вы указывали в начале.

Все вот так легко и просто сделать коммит в основной репозиторий, потом через интерфейс github можно сделать мердж в главную ветку main.

Пользуйтесь! 😄🤗

#Интрересная история про GIT.
Как связать локальный репозиторий с удаленным сервером и ничего не сломать 🧞

Git был создан Линусом Торвальдсом (создателем ядра Linux) в 2005 году.
Причина - конфликт с владельцами проприетарной системы контроля версий BitKeeper, которую использовали для разработки ядра Linux.
Линус захотел:

- Полностью распределённую систему (каждый разработчик имеет полную копию репозитория).
- Очень быструю работу.
- Защиту от подмены истории (Git использует SHA-1/2 хэши).
- Надёжное управление ветками и слияниями.

С тех пор Git стал де-факто стандартом в мире разработки ⚠️

- Локальный репозиторий - находится у тебя на компьютере.
- Удалённый репозиторий - размещён на сервере (например, GitHub, GitLab, Bitbucket) и используется для совместной работы.

Что хранится в репозитории:
- Файлы проекта (код, конфиги и т.д.)
- История изменений (git log)
- Ветки разработки (git branch)
- Метаданные Git (например, .git папка)
- Твои обнаженные фотки 🤫

В видео примере показал как на практике использовать один и тот же .git в разных местах 🔗

Что может быть лучше чем финальное видео про Airflow и настройку
`airflow.cfg`
пятничным вечером?

Да что угодно лучше этого 😆

Но нельзя просто так взять и не завершить базовую настройку Airflow для дальнейшей игры в дата инженеров(мамкиных).
`airflow.cfg`

Это буквально сердце Airflow в котором есть набор самых важных параметров необходимых для того чтобы он отдавал вам всю мощь.🌿
Давайте посмотрим на популярные настройки которыми нельзя принебречь:

ВСЕГДА используйте UTC. Если у вас возникнет вопрос почему, ответ простой: "потому". Сами напишите в комментариях ответ 😁

default_timezone = utc

executor - Из прошлого видео вы уже знаете какие бывают.

executor = LocalExecutor

Параллелизм - максимальное число запущенных одновременно задач во всех дагах(включая сами даги).
Макс таск пердаг - максимальное одновременное количество активных задач в даге.
Макс актив ранс пердаг - максиальное количество запущенных ранов 1 дага.

parallelism = 16

max_active_tasks_per_dag = 4

max_active_runs_per_dag = 8

Последние три настройки базируются исключительно на вашей железке и способе деплоя.
Красивый пример как считать как эти параметры настраивать:
parallelism = ваши cpu * 2 очень грубо, но это близко к правде. Как вы успели заметить на наших 4 cpu ни о каких 16ти речи идти не может. Но мы люди творческие че хотим то и делаем 😆

max_active_tasks_per_dag = тут просто сколько одновременно можно задач в даге делать. Это важно, чтобы один DAG не занял всё. У нас маленькая тачка 4 вполне оптимально. Да и сам даг подразумевает некую последовательность в выполнении тасок, но парарлелить их тоже можно 😠

max_active_runs_per_dag = ну 8 для нашей машины много, оставим потом 2-3. Суть в том что если один запуск еще не закончился а расписание подошло то таких случаев 8 к ряду можно сделать, ситуация очень редкая и такое вообще допускать плохая практика. Однако это роляет при каче исторических данных за большой период. Пока что для нас параметр бесполезный в целом, но его тоже важно уметь настраивать.

🥳 Что ж поздравляю вы дошли до финиша этого образотаельного трэка дальше будем упражняться это самое веселое 🤪

#DEMeme

Вся правда в одной картинке 📣

#DEMeme

За каждым дашбордом свой проект «Манхеттен»
Мем смешной ситуация не простая 📃✂️🪨

#PyTrickMonday

dict.get() : маленькая таблетка от KeyError-ов

Вы уверен знаете что из словаря можно взять значение по ключу несколькими способами

payload = {"user": {"name": "Alice"}}

# Явное обращение по ключу
nickname = payload["user"]["nickname"]          # KeyError, DAG падает

# Спокойный вариант c .get()
nickname = payload.get("user", {}).get("nickname", "Anon")

# "Если ты сигма" - способ (когда поле может отсутствовать целиком)
nickname = (payload.get("user") or {}).get("nickname", "Anon")

Так вот я вам предлагаю использовать .get() ВСЕГДА и у меня на это пять три причины:

1️⃣Безопасность: API вернул 200 OK, но ключа "user" нет - прямое обращение бросит KeyError и уронит скрипт. .get() вернёт None (или ваш дефолт) и программа продолжит работу.

2️⃣Меньше грязного кода: вместо громоздкого: (Плоское лучше вложенного ZenOfPython)

if "key" in resp:
    value = resp["key"]

# одна строчка
value = resp.get("key")

3️⃣Гибкий дефолт: jobs.get("DA", "DE") # вернёт 'DE', если ключа 'DA' нет

💡 Если надо "нырнуть" глубоко и не облажаться по пути, так как .get есть только у словарей, можно дефлтом указывать пустой словарь и ваш код не упадет.
email = payload.get("user", {}).get("contacts", {}).get("email")

📣 Отправь этот полезный пост коллеге чтобы он/она не ловили KeyError и были тебе всегда благодарны 😎

#DETip #Python #DataJungle

#SQLWednesday
Как же на самом деле крутится "SELECT" 🤪

Как же "движок" базы данных обрабатывает ваш запрос в каком порядке.
Когда то на собеседовании на джуновскую позицию мне задали такой вопрос, может и вам пригодится. И кстати может вы избавитесь от ошибок в текущей работе ⚠️
Логический порядок такой:
FROM → JOIN → WHERE → GROUP BY → HAVING → WINDOW → SELECT → DISTINCT → ORDER BY → LIMIT/OFFSET

Давайте поближе поглядим что происходит на каждом этапе(Да не все запросы содержат ВСЕ операторы но порядок в любом случае такой):

1. FROM --Сначала SQL надо сказать откуда мы берем данные из какой таблицы.
2. JOIN --Логично если на первом месте таблицы то JOIN точно на втором.
3. WHERE --Фильтруем строки после соединения. Теперь понимаете почему нельзя использовать алиасы из SELECT, потому что он только 7 на очереди👀
4. GROUP BY --Лепим группы, считаем агрегаты.
5. HAVING --Фильтруем уже сведённые группы.
6. WINDOW --Считаем оконные функции (ROW_NUMBER, avg() over…). Они видят итог после WHERE, но до SELECT (уверен вы сейчас такие ЧЕ??)
7. SELECT --Выбираем финальные столбцы, присваиваем алиасы. Только сейчас ⚠️
8. DISTINCT --Убираем дубликаты.
9. ORDER BY --Сортируем. Уже видим алиасы из SELECT.
10. LIMIT/OFFSET --Отрезаем кусок результата.

❗️Ставь 🔥 если не знал что порядок именно такой 👀

Давайте примеры посмотрим:

-- Сколько авторов имеют > 3 статей
SELECT   author_id,
         COUNT(*) AS article_cnt
FROM     articles
GROUP BY author_id
HAVING   COUNT(*) > 3;

1. FROM - берём articles
2. GROUP BY - группируем по author_id
3. HAVING - фильтруем готовые группы (в WHERE так не выйдет - агрегатов ещё нет)
4. SELECT - выводим автора и число статей

-- Хотим все ордера + успешные доставки
SELECT  o.id,
        d.tracking_code
FROM    orders o
LEFT JOIN deliveries d
  ON    d.order_id = o.id
  AND   d.status = 'shipped'   -- фильтр СРАЗУ при соединении
WHERE   o.created_at >= current_date - interval '30 days';

Фильтр в JOIN … ON ускоряет LEFT JOIN
Предикат status = 'shipped' выполняется до WHERE, поэтому пустые доставки всё ещё вернутся как NULL, но оптимизатор не тащит лишние статусы.
Это один из способов как можно ускорить ваш запрос вынося из WHERE в JOIN.

WITH active AS (
  SELECT *
  FROM   sessions
  WHERE  started_at >= current_date - 7
)
SELECT  user_id,
        started_at,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY started_at) AS rn
FROM    active
ORDER BY user_id, rn;

Оконка «видит» фильтр из WHERE, но идёт раньше SELECT
WHERE уже отрезал старые сессии.
ROW_NUMBER считается, потом идёт SELECT, где мы выводим rn.

Красиво не правда ли? 😃
Пересылайте коллегам и друзьям шпаргалку по SQL сохраняйте себе в телеграм 😮


Давайте в комментариях обсудим как улучшить этот запрос:

-- Ищем заказы c дорогими товарами и уже оплаченным счётом
SELECT  o.id,
        o.created_at,
        c.total_amount,
        c.status    AS charge_status,
        p.name      AS product_name,
        p.price
FROM    orders      o
JOIN    order_items oi  ON oi.order_id   = o.id
JOIN    products    p   ON p.id          = oi.product_id
JOIN    charges     c   ON c.order_id    = o.id
WHERE   p.price      > 1000                -- фильтр по цене
  AND   c.status     = 'paid'              -- фильтр по чеку
  AND   o.created_at >= current_date - interval '90 days';

За лучший ответ получите статус "Порядочного" в чате DE Data Talks 🙂

Ну что сегодня пятница а значит время продолжить большой разбор #Airflow 🚀

Сегодня в большом видео примере вы узнаете:

1. Как писать свой первый #DAG.
2. В чем разница между requests.get() и requests.Session().get() она есть и приличная.
3. Напишем на #python красивый класс для API #CoinMarketCap.
4. Запустим наш даг, получим ошибку и еще разок запустим 😎


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

Пересылайте друзьям и коллегам кто хочет научится писать свой первый даг но не знает с чего начать 📣

Любые вопросы в комментариях, всегда к вашим услугам.

В следующем видео сделаем #CICD и установим #Postgres на наш сервер чтобы там хранить данные не пропускайте! 🙄

#DEmeme

Сколько дашбордов не делай, а бизнесс все равно в эксель смотрит 😶

#DEMeme
Слово AI из каждого утюга, возможно даже сам утюг под собой скрывает нейросеть ⚙️

В этом меме много боли на самом деле 😄

#PyTrickMonday

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

Уверен вы прекрасно знакомы с конструкцией with open(file, 'r') as file_to_open:
Или при работе с базой with engine.connect() as conn:
А вот как сделать свой собственный контекстный менеджер? Сейчас все покажу! 💡

Для начала что такое менеджер контекста with?
Это такой специальный объект имеющий два магических метода - магия в том что они прячутся за словом with и вы их не видите 👀

__enter__()   # запускается при входе в with-блок  
__exit__()    # вызывается при выходе (даже если внутри случился Exception)

Для примера давайте перепридумаем функцию open()

from contextlib import contextmanager # импорт контекстного менеджера

@contextmanager
def open_file(path, mode="r"):
    print("открывашка")
    f = open(path, mode)
    try:
        yield f # всё, что до yield это __enter__; все что после это 

ачалом нов

    finally:
        print("закрывашка")
        f.close()


with open_file("send_log.txt") as f: # наш собственный контекстный оператор
    data = f.readline()
    print(data)

# Вывод:
открывашка
2025-05-30 16:24:14,565 | INFO | Connecting to 149.154.167.91:443/TcpFull...
закрывашка

Самое важное что тут надо понять что там где yield - это то что вы делаете в теле контекста 😃

Давайте что то более жизненное разберем.
Например "Я хочу знать сколько памяти мой датафрейм съел после открытия файла большого"
Вот как это можно красиво сделать:

import os, psutil
from contextlib import contextmanager

@contextmanager
def mem_used(name: str = "block"):
    proc   = psutil.Process(os.getpid()) # берет текущий процесс
    start  = proc.memory_info().rss # размер в bytes
    try:
        yield # вот тут вы откроете ваш файл
    finally:
        end   = proc.memory_info().rss # было памяти у процесса в начале
        diff  = end - start  # сколько съело ваше чтение bytes
        mb    = 1024 ** 2                     # bytes в MiB

        print(f"MemoryUsed: {diff/mb:+.2f} MiB")


with mem_used("parse_json"):
    big_df = pd.read_json("og.json", lines=True, orient="records")

# MemoryUsed: +1.17 MiB

Пересылайте коллегам, чтобы они наконец-то они узнали что же за магия скрывается за WITH 🪄

Теперь домашнее задание напишите контекстный менеджер считающий время выполнение вашего кода with timer("любое имя процесса")
вывод должен быть такой "Имя процесса: время в секундах"
Присылайте решения в комменты 🔽

Делаем сами без помощи GPT! в этом посте все есть, чтобы вы справились, разогрейте мозги в начале недели 🧠

2️⃣🔤🔤🔤🔤🔤 без ИИ #DEInsight

ИИ технологии настолько сильно проникли в мою жизнь что уже и не припомню когда бы был хоть день, когда я не задавал какой то вопрос GPT.
И мне стало интересно "А что если я на работе вообще не буду пользоваться ИИ"
Вот что я понял за 2 недели, работая без ИИ помошников(GPT, co-Pilot). Будет душновато отркываем форточки 😄

0️⃣Все ответы есть в ДОКУМЕНТАЦИИ или в Reddit или в StackOver. Читайте доку и вам не придется пытать и оскорблять ГПТ пока он не даст ответ 🙂
1️⃣Получая ответ от ГПТ я просто использовал, не запоминая особо, а через неделю мог спросить тоже самое легко, но прочитав доку, поресерчив в интренете и найдя ответ он остался в моей памяти и за две недели я освежил много уже забытых знаний и получил кое что новое. Я это "выстрадал" поэтому оно отпечаталось.
2️⃣Использование ИИ создает ложное ощущение что твоя экспертиза растет ведь ты можешь сделать гораздо больше. Но это иллюзия и самообман, когда ты не провел ресерч должным образом, а просто выбрал "думающую" модель для лучшего ответа она сама за тебя почитала доку и дала тебе решение, твой вклад в это только задданый вопрос.
3️⃣Чтобы запоминать и получать опыт надо иметь взаимосвязь: Выявление проблемы - Поиск ответа(чтение доки, редита) - Использование найденных ответов -- после нескольких итераций таких -- Находишь решение и объяснение этому решению. И весь этот процесс ресерча создает прочные нейронные связи и это тот же дофамин но более сложный. А ИИ за вас делает ресерч и выдает решение и когда у "вас" получилось вы тоже получаете дофамин, но быстрый - и это проблема.
3️⃣🔤1️⃣ Проведите аналогию между съесть шоколадку и полежать дома сейчас чтобы стало хорошо, или продолжать следить за питанием и заниматься спортом чтобы отлично выглядеть но потом.
Вы получите удовольствие в обоих случаях, но во втором это более сложный путь и он скучнее, но для вашего здоровья и фигуры гораздо более правильный выбор да и опыт который вы получите куда более применим чем шоколад и диван 🙂
Использование ИИ в решении ваших рабочих(любых) задач это таже ловушка и с этим надо быть аккуратнее.
4️⃣Есть иллюзия что ИИ помогает быстрее решать задачи, это не правда. Скорость только в начале упала, но "поскрипя" пару дней быстро восстанавливаешься. Потому что то что раньше делалось на автомате, потом делегировалось ИИ и поэтому пришлось "поскрипеть".
5️⃣У меня вновь появилось желание изучать новое, оно как будто бы заснуло на время.
6️⃣В первые два дня я не мог написать рекурсивную функцию примерно час 😄 А раньше такое делалалось изи. Я мог и не писать ее, но мне хотелось.
7️⃣Критическое мышление с ИИ притупляется - факт. Обычно у вас так - "ой че то он херь пишет, сам сделаю". А это уже звоночек что, не получая рузультата, такого как надо вам вы уже не тратите время на осмысления, почему он пошел не туда. Или может вы его туда завели 😄просто потому что не углубились в проблему которую решаете.
8️⃣Английский. Я так ленился сам читать и переводить и писать, что давал ГПТ и говорил ответь вот что и просто копировал отсылая обратно. Поделав это самостоятельно вновь, я понял что опять же у меня была иллюзия что это "Я" общаюсь, потому что я писал в промпт что надо перевести на английский. Но это вообще не я и это не мои слова я просто в очередной раз просил ГПТ решить задачу за меня. Даже не техническую.


Вывод какой напрашивается:
ИИ это прекрасно, но надо делать передышку и вернуть ресерч в вашу жизнь. Иначе вы буквально тупеете, а думаете что эксперт 🤷‍♀️
После пары недель я чувствую прилив знаний и дофамина от решения задач самостоятельно.
Быстро понимаешь где у тебя реальные знания есть(или были) а где ГПТ-завеса.
Когда думал что экономил время на самом деле экономил мозги.


Такие философские рассуждения.
Это лишь мое мнение - мой опыт(НЕ ИСТИНА), вы можете думать иначе, пишите в коментариях что думаете 🪄

#SQLWednesday

Сегодня обсудим классический вопрос с собеседования для #DataEngineer. 🪄
Вопрос звучит так:
"Перед тобой #csv файл 20ГБ, как откроешь?"
Конечно никакой pd.read_csv() тут не поможет это просто уложит ваше ядро в #JupyterNotebook.
-- Или поможет ? Дочитай до конца и узнаешь.
Такой вопрос могут задать на самом деле и Junior и Middle да и выше. Потому что не всегда очевидно как это сделать красиво и эффективно.

Самый эффективный способ работать с такого размера файлом это естесвенно загрузить его в базу.
#Postgres уже имеет нативный инструмент для работы с csv файлами и вот как это выглядит.

COPY raw.sales
FROM '/var/lib/postgresql/big_sales.csv'   -- путь до файла на сервере или локальный путь(если постгре локальная).
WITH (
    FORMAT      csv,       -- CSV-парсер
    HEADER      true,      -- пропускаем строку заголовков
    DELIMITER   ',',       -- если вдруг не запятая
    QUOTE       '"',       -- кавычки по умолчанию
    ENCODING    'UTF8'     -- кодировка можно изменить если будет "криво"
);

Легко и просто. Но мы не всегда имеем доступ к серверной файловой системе, так вот как это сделать прям из JupyterNotebook не ломая ядро?

1. Для нашего эксперимента я сгенерировал файл гигант вес которого 13.4GB
2. У меня уже есть развернутый постгрес на удаленном сервере(вы можете локально установить)
3. Будем пользоваться любимым пандас с chunksize

import pandas as pd
import sqlalchemy as sa

DB = dict(
    user="",
    password="",
    host="",
    port=5432,
    dbname="",
)
engine = sa.create_engine(
    f"postgresql+psycopg2://{DB['user']}:{DB['password']}@{DB['host']}:{DB['port']}/{DB['dbname']}",
)

#  параметры CSV
csv_path   = "big_file.csv"
chunk_rows = 100_000 # будем грузить по 100K строк
table_name = "example_csv"
schema     = "raw"

# создаём таблицу один раз с нужными колонками
with engine.begin() as conn:
    conn.exec_driver_sql(f"""
        CREATE SCHEMA IF NOT EXISTS {schema};
        DROP TABLE IF EXISTS {schema}.{table_name};
        CREATE TABLE {schema}.{table_name} (
            col_0 text, col_1 text, col_2 text, col_3 text, col_4 text,
            col_5 text, col_6 text, col_7 text, col_8 text, col_9 text
        );
    """)

# грузим чанками
reader = pd.read_csv(
    csv_path,
    chunksize=chunk_rows,
    iterator=True,
    header=0,
)

for i, chunk in enumerate(reader, 1):
    chunk.to_sql(
        name=table_name,
        con=engine,
        schema=schema,
        if_exists="append",
        index=False,
        method="multi",
    )
    print(f"Чанк {i}: {len(chunk)} строк залито")

print("Готово!")

Пересылайте коллегам, чтобы не терялись на собеседованиях и знали что ответить 💡

Чтобы поиграть дома в DE тетрадку с кодом прилагаю 🧠

#DataJungle #Postgres #ETL

Сегодня раворачиваем #Postgres, настраиваем #CICD, выходим из #vim !🤝

Это финальное видео в серии о том как деплоить #Airflow и делать свои даги.
Как и обещал рассказал про CICD и о том как развернуть postgres, чтобы наши данные было бы где хранить.

Как разворачивать postgres в #ubuntu как настоящий девопс?

1. Устанавливаем саму базу:

sudo apt install postgresql

2. Разрешаем слушать внешние адреса файл ( /etc/postgresql/*/main/postgresql.conf ):

 
listen_addresses = '*'

3. Разрешаем коннект именно с вашего IP(файл /etc/postgresql/*/main/pg_hba.conf ) на нижней строке добавляем.

host DB_NAME       USSERNAME        YOUR_IP        md5

4. Заходим в CLI psql создаем базу и пользователя:

sudo -u postgres psql

CREATE DATABASE DB_NAME;
CREATE USER USERNAME WITH PASSWORD 'PASSWORD';

Все на этом базовый деплой базы будет завершен. Все остальные подробности в видео!

Атракцион невиданной щедрости! 🔔
И самое главное, я обещал что вы поиграете в реального #DataEngineer что для этого надо:
1. Быть подписанным на классный канал про дата инжиниринг @data_jungle.
2. Быть в чате этого канала.
3. В комментарии к этому посту написать "хочу поиграть в DE ВАШ НИК В ГИТХАБ и ВАШ IP ноута с которого работаете".

После этого вы получите
- инвайт в репозиторий
- крэды в Airflow
- крэды в БД на ЧТЕНИЕ(это значит что инсерт вы сможете сделать только через ДАГ, а читать можете с jupyter из дома).

Это позволит вам со своей ветки пушить код в наш скромный продакшн 👀 я буду лично проводить ревью кода и писать комментарии чтобы вы могли становится лучше.

Правила игры:
- нельзя использовать метод pandas.to_sql все делаем только через хук как на видео, предварительно создавайте таблицы через таску в #airflow в схеме raw_data.
- данные можете качать какие угодно, любой #API который нравится, просто добавляйте ваш ключ в переменные и пользуйтесь.
- нельзя хранить больше 1М строк в вашей таблице в базе.

Не бойтесь уронить Airflow, вы учитесь это нормально, задавайте вопросы не стейсняйтесь, и да прибудет с вами сила 🔫

#PyTrickMonday

С началом новой недели друзья, сегодня 3 нереально крутых приёма в #pandas, про которые большинство даже не догадываются 👀

1️⃣Series.explode() - превращаем «списки в ячейках» в длинную таблицу одной строчкой. Очень полезная штука если например у вас есть вот такой объект когда к 1 заказу есть два айтема и вы бы хотели получить такой вид:
order_id items
0 101 BTC
1 101 ETH
2 102 DOGE

import pandas as pd

orders = pd.DataFrame({
    "order_id": [101, 102],
    "items": [["BTC", "ETH"], ["DOGE"]]
})
flat = orders.explode("items", ignore_index=True)

Просто используйте метод explode() для вашей колонки с вложенными данными ✔️

2️⃣ #DataFrame .pipe() - конвейер #DataEngineering по взрослому.

Pandas любит цепочки вызовов вместо лапши кода. Если вам надо применить к датафрейму много функций то можно выстроить это в pipeline очень легко и просто.

def add_profit(df):
    return df.assign(profit=df.sales - df.cost)

def top(df, n=5):
    return df.nlargest(n, "profit")

df = (pd.read_csv("shop.csv")
        .pipe(add_profit) # добавляем новую колонку профит
        .pipe(top, n=3))  # берем топ 3 по профиту

# передаёшь прямо в pipe свои аргументы функций

Очень удобно, меньше кода больше эффективности ✔️

3️⃣DataFrame.convert_dtypes() - минус RAM в одну команду.
Загружаешь CSV - и видишь большинство типов object и для дальнейшей работы надо делать преобразование типов .astype().

raw   = pd.read_csv("huge.csv")
clean = raw.convert_dtypes()

# переводит «строки» в современный тип string[pyarrow] или string
# меняет int64 на nullable Int64, NaN больше не проблема
# падает объём памяти до ×4 раз (часто - и больше)

Одна строчка экономит кучу памяти ✔️

Напиши в комментариях сколько удалось сэкономить места с .convert_dtypes() проверь на каком то жирном DataFrame🧐

Пересылай друзьям и коллегам, уверен они будут благодарны тебе 👀