NEW BOT Телеграм, страница

Управляем ошибками. Часть 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

Please open Telegram to view this post