"Управляем ошибками. Часть 2 📼 Late Data Посмотрим на 3 паттерна для работы с данными, которые пришли позже, чем ожидалось 1️⃣Pattern: Late Data Detector Пока что для меня сложная и непонятная история про стриминг. В целом, книга легко читается, но требуется время, чтобы переварить. Все концепты сжаты, но очень насыщенны. Перечитаю, когда нужно будет с этим работать Интересная мысль - ""shifting the late data problem"" Пример: мы пишем партиции по времени обработки. В партицию 21:00 к нам залетел кусок данных за 20:00 и 19:00. А наши пользователи используют партиции по времени события. Тогда мы перекладываем ответственность ковыряться в этих партициях на них 😁 2️⃣Pattern: Static Late Data Integrator Как вообще можно перегрузить данные за прошлое? 1. Создать кучу дагранов, где каждый перегружает 1 день. Если упало - перезапускаем конкретный день 2. Создать один дагран, где в коде генерируется список нужных дат. И по каждой дате запускается загрузка. Если упало - просто перезапускаем, пойдет считаться с упавшего дня. Это и есть Static Late Data Integrator. А статическое - потому что мы сами задаем 14 дней или сколько угодно И тут я поняла, что неосознанно это и делала. У нас часто была проблема, что данные в источники просто не приходили 😁 Потом мы шли разбираться с владельцами, и данные заливались, но позднее. Чтобы это учитывать, в моем подходе был такой алгоритм: 1. Задаем стартовую и конечную даты расчета 2. Создаем диапазон значений ``` full_range = pd.date_range(start=str(start_dt), end=str(end_dt)).strftime(""%Y-%m-%d"") ``` 3. Из меты достаем существующие партиции ``` def get_existing_partitions(table_name): partitions = ( spark.sql(f""show partitions {table_name}"") .select(F.split(F.col(""partition""), ""="")[1].alias(""dt"")) .collect() ) return [p[0] for p in partitions] ``` 4. Находим разницу ``` lost_range = full_range.difference(existing_partitions_pdf) ``` 5. Итерируемся по потеряшкам ``` for dt in lost_range: calc_mart(dt) ``` Если в будущем снова будет пустая дата, нам не придется перезапускать определенный день - он пойдет считаться сам 3️⃣Pattern: Dynamic Late Data Integrator Предлагается завести табличку с 4 полями: 🤩партиция 🤩время обработки 🤩время добавления новых записей 🤩флаг обработано или нет Так мы запросом можем найти партиции, которые уже обрабатывались, но в которые попали новые данные. А в iceberg есть удобное свойство last_updated_at на уровне таблицы 🤩 Filtering Pattern: Filter Interceptor Как будто это антипаттерн. Предлагается создать доп колонки с фильтрами id_is_not_null, status_is_not_failed и выводить количество отфильтрованных записей, чтобы понимать, на каком этапе ошибка в коде или в данных. Но прям пробегаться по каждой записи в датафрейме… Как будто это все-таки dq 🌳 Fault Tolerance Pattern: Checkpointer Просто нужно создавать чекпоинты и хранить последний оффсет обработанной записи и состояние, если оно есть Еще раз напомнили про семантики доставки: 🤩exactly once - нужны другие паттерны, расскажу, когда дойду 🤩at least once - чекпоинт после обработки, могут быть дубликаты при перезапуске после падения 🤩at most once - чекпоинт до обработки, данные потеряются при падении #depatterns"