Расскажу про реальную задачу на работе. Сразу скажу, что это не хай лоад перфоманс дата сайенс задача. Поэтому, если ты заряженный спец или спецуха, то наши с тобой пути расходятся. Дело не в тебе, дело во мне. Я больше не люблю тебя.. Твоя мама была права.
Как дают задачу?
Возьми этот CSV файлик, там фразы и слова, которые люди ищут в поиске Яндекса. Надо пройтись по всем запросам и собрать инфу из Яндекс Вордстата, сколько раз люди ищут это и конечный результат-отчет загрузить в Clickhouse. Например "Как заработать на беху купешку" вчера искали 3 раза, а "Я хочу ничего не делать и кататься на бехе" искали 4530 раз.
И с чего начать?!
Конечно же с чата гпт. Ведь ИИ уже заменил программистов и теперь я вообще ничего не делаю и просто пью коктейль на берегу океана. Было бы круто, но не совсем так. Код надо писать почти с нуля, причем сразу в общем виде, чтобы его можно было переиспользовать на следующих задачах. Поэтому сразу прикидываю архитектурно, как это может выглядеть:
1. API - выгружаю при помощи python и сохраняю в S3, как есть без трансформации
2. Забираю из S3 и кладу в Clickhouse с помощью Spark (тут уже сделаю преобразования)
3. Пишу DAG в Airflow с двумя тасками (и хочу сделать ручной запуск со своим конфигом)
В целом изи. Че тут писать. Но есть некоторые нюансы. В ТЗ мне указали, что нужны недельные данные. Это значит, что при обращении к АПИ без уважения мне надо тянуть диапазон дат (НЕДЕЛЮ). Ну вы скажите, ок, запускай каждый понедельник и тяни данные за прошлую неделю. И ТУТ ТОЖЕ НЕ ПОЛУЧИТСЯ.
Яндекс Вордстат отдает данные минимум за ТРИ недели. Поэтому либо каждый третий понедельник запускать, либо думать по-другому. Я выбрал второе (Хотя думать я не привык) Буду запускать каждый понедельник, чтобы не сидеть без данных несколько недель. А скаченные данные сохранять в папки с названием api_wordstat_2025-10-20_2025-11-09. Ну т.е. у меня будут сохраняться папки с диапазоном в три недели. И пусть данные в S3 перескаются — это все потом фиксанет Clickhouse (причем без меня сам).
Ок, вроде первый пункт готов. Но НЕТ! А откуда мне и как забирать сами фразы-то для API? Поэтому я загрузил CSV с фразами в Clickhouse и написал класс на python с подключением через clickhouse_connect к CH. Почему класс? А чтобы потом его можно переиспользовать для других задач. Там методы на чтение, запись и кастомного запроса. Помните, что у меня изначально не было проекта. Писал все с нуля.
Итак. Я подключился к CH, забрал фразы, отправил их в API, получил ответ, сложил в S3. А что насчет фраз, которые содержат всякие невалидные символы, типа —> % ~§ и так далее? Тут я сделал логику, что в случае нахождения плохих запросов, записывать их в отдельную таблицу Clickhouse вместе с текстом ошибки (Помните мой ранее написанный класс на питоне? Вот вот). Тем самым я не роняю расчет, а просто логирую это недоразумение и иду дальше. Это кстати работает, если даже у меня соединения внезапно пропало. Все логируется и можно отследить.
А ну еще. Есть у меня логика всяких там retry. Типа если один раз не получилось именно подключиться, то сделай паузу, потом еще одну, но чуть подольше и только потом иди дальше.
ОКЭЙ. Дальше пишем класс для загрузки из S3 в CH. Там у меня три метода. Вы спросите: "ТАК НЕЛЬЗЯ ЦКАТИНА". Короче, я заметил, что если просто спарком читать данные из клика, то он все записывает на один executor, поэтому написал отдельный метод, который читает хитрым способом несколькими частями и размазывает по всему спарку. Не буду вдаваться в подробностями. Мб в роадмап загружу пример.
Че по ДАГу жи есть?
Сделал три таски
• compute_dates определяет диапазон (по умолчанию — последние 3 недели),
• api_to_s3 грузит данные из api в s3 через python
• s3_to_ch загружает готовые parquet-файлы в ClickHouse.
• Между тасками у меня XCom, где я передаю даты из первой таски, чтобы брать конкретные названия папок для Spark
Че по дублям. У меня же там накладываются недели друг на друга, когда берешь данные. А мне пофигу, я настроил ReplacingMergeTree в кликхаусе. Он и дедублицирует. Хотите еще?