"Крутим ~~казик~~ API Wordstat! Расскажу про реальную задачу на работе. Сразу скажу, что это не хай лоад перфоманс дата сайенс задача. Поэтому, если ты заряженный спец или спецуха, то наши с тобой пути расходятся. Дело не в тебе, дело во мне. Я больше не люблю тебя.. Твоя мама была права. Как дают задачу? Возьми этот 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 в кликхаусе. Он и дедублицирует. Хотите еще?"