В источник добавили столбец: как отличить безопасное изменение схемы от поломки

В выгрузке заказов появился столбец promo_code. Старый загрузчик выбирает поля по позициям: третий столбец считает суммой. Если новый столбец добавили перед суммой, деньги превращаются в текст купона. Если добавили в конце, всё может продолжить работать. Оба изменения называют «добавили поле», хотя последствия определяет договор между источником и потребителем.

Чтобы отличать безопасную эволюцию схемы от поломки, опишите требования читателя. Какие поля обязательны, какие могут отсутствовать, какие допускают null, а какие дополнительные поля разрешено игнорировать. Затем сравнивайте каждую новую запись с этими требованиями. Совместимость — свойство конкретного потребителя и направления обмена, а не названия изменения.

Отсутствие поля и null — разные состояния

Необязательный promo_code может совсем не приходить для старого источника. В новой записи ключ может присутствовать со значением null, означающим отсутствие купона. Эти случаи удобно приводить к одному внутреннему представлению, если бизнес-смысл совпадает. Но обязательная сумма с null требует отказа: поле существует, пригодного значения нет. Проверка только наличия ключа эту ошибку пропустит.

В JSON Schema перечисление properties само по себе не делает поле обязательным: для этого существует required. Неизвестные свойства тоже имеют отдельное правило. Даже если вы пока не используете библиотеку валидации, полезно явно воспроизвести эти три решения в контракте и тестах.

Проверим четыре изменения

В примере потребитель требует order_id как строку и amount_cents как целое число. Значение хранится в копейках, чтобы тип не зависел от десятичного разделителя. Купон необязателен и может быть null. Новое поле source разрешено: потребитель его игнорирует. Проверка типа через type нужна здесь ещё и потому, что bool в Python является подклассом int.

def validate(row):
    problems = []
    required = {"order_id": str, "amount_cents": int}
    for key, kind in required.items():
        if key not in row:
            problems.append("missing:" + key)
        elif type(row[key]) is not kind:
            problems.append("type:" + key)
    if "promo_code" in row and row["promo_code"] is not None:
        if type(row["promo_code"]) is not str:
            problems.append("type:promo_code")
    return problems
cases = [
    {"order_id": "A", "amount_cents": 1200, "source": "app"},
    {"order_id": "B", "amount_cents": 800, "promo_code": None},
    {"order_id": "C", "amount_cents": "800"},
    {"order_id": "D"},
]
result = [validate(row) for row in cases]
print(result)
assert result == [[], [], ["type:amount_cents"], ["missing:amount_cents"]]

Добавление source не повредило этому читателю, а превращение суммы в строку нарушило контракт. Автоматически привести "800" к числу иногда допустимо, но это должно быть согласованное преобразование с отчётом об ошибках. Иначе тот же путь однажды примет "8,00" как восемьсот либо отвергнет половину источника после смены локали.

Почему добавление обязательного поля опасно

Если новый потребитель требует currency, старые сообщения его не содержат. Сначала выпустите читателя, который понимает прежнюю схему и новый необязательный атрибут. Затем обновите источник. Делать поле обязательным можно после проверки, что все актуальные отправители его заполняют и история обработана по явному правилу. Для старых заказов валюта может восстанавливаться по магазину и дате; случайный общий default способен испортить аналитику.

Обратное направление тоже проверяйте: старый потребитель должен пережить новые записи, если они ещё будут к нему попадать. При запрете неизвестных полей добавление даже необязательного атрибута станет нарушением. Такой строгий режим бывает полезен для обнаружения опечаток, но требует согласованного обновления участников. Выберите правило исходя из способа выпуска ваших сервисов.

Для табличного входа перестаньте связывать смысл с порядком столбцов, если контракт задаёт имена. Сначала проверьте заголовок, затем выберите необходимые поля по имени. Отдельно контролируйте дубли имён: два amount_cents делают выбор неоднозначным. Семантическая смена единиц с рублей на копейки также требует новой версии, даже когда имя и технический тип остались прежними.

Сохраните по примеру записи каждой поддерживаемой версии и запускайте их через настоящий загрузчик перед релизом. В отчёте укажите принятые, отклонённые и преобразованные строки. Так изменение источника превращается в проверяемую миграцию, а не в спор о том, считается ли добавленный столбец безопасным.

При добавлении обязательного признака проверьте ещё архивное повторное чтение. Сегодня все отправители уже обновлены, но backfill поднимет вчерашние сообщения без нового поля. Политика миграции истории должна быть такой же явной, как обновление текущего потока.

В курсе разобраны схемы и контракты сообщений, batch и streaming, Kafka, Spark и Airflow. Следующий полезный шаг — встроить проверку совместимости в конвейер и довести новые и старые записи до общей таблицы с понятным отчётом качества. Откройте программу платного курса «Data Engineering Fast‑Track: Kafka → Airflow → Spark».