LatornLatorn
Войти

CDC в pandas: идемпотентный MERGE с проверкой

Одна и та же пачка изменений может прийти повторно. Напишите обработчик CDC на pandas, который применяет вставки, обновления и удаления, сохраняя результат при повторе той же пачки. Схема входных таблиц и правила выбора события приведены в условии.

Продолжить практику с Kafka, Airflow и Spark можно в платном курсе Data Engineering Fast-Track.

Условие

Задача. Реализуйте идемпотентное применение порции CDC к «серебряной» таблице.

Вам даны два pandas.DataFrame:

  • silver: текущее состояние таблицы
    колонки: id: str, name: str, price: float, updated_at: datetime

  • cdc: пакет изменений (может содержать несколько событий на один id, порядок не гарантирован)
    колонки: op: {'I','U','D'}, id: str, name: str?, price: float?, event_ts: datetime, source_lsn: int

Нужно:

  1. Нормализовать пакет CDC: для каждого id оставить последнее событие по (event_ts, source_lsn) (latest-wins).

  2. Применить MERGE:

    • op='D' → удалить строку с id из silver.

    • op in {'I','U'} → upsert: если id есть — обновить; если нет — вставить.

    • updated_at = event_ts.

    • Если в CDC для поля name/price значение отсутствует (NaN), сохранить старое значение (не затирать NaN).

  3. Вернуть кортеж (silver_next, plan_str), где plan_str содержит метку MERGE_OK.

Требование по идемпотентности: повторное применение того же cdc к silver_next не должно менять результат.

Сигнатура:

def apply_cdc_merge(silver: pd.DataFrame, cdc: pd.DataFrame) -> tuple[pd.DataFrame, str]:
    ...
РешениеPython

Без регистрации. Гостевая проверка не сохраняет прогресс. Для сохранения перейдите к уроку.

Отправьте решение, чтобы увидеть результаты тестов.

Связанные задачи