Проверка схемы и утверждения

Pipeline автоматически обрабатывает новые данные, часто без участия человека. Если схема входного файла изменится — столбец будет переименован, формат даты изменится или появится новая категория, — pipeline должен немедленно завершиться с ошибкой, а не создавать незаметно неправильный результат. Проверка схемы с помощью утверждений обеспечивает соблюдение соглашений между поставщиками и потребителями данных и позволяет обнаружить проблемы как можно раньше.

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

После проверки обязательных столбцов проверьте их типы данных. Если столбец с датами загружен как object, это означает, что не был вызван pd.to_datetime(). Если столбец с идентификаторами хранится как float, а не как int64, часто это указывает на наличие значений NaN, из-за которых невозможно хранить данные как целые числа. Для наиболее важных столбцов задавайте проверки типов с помощью assert df['col'].dtype == expected_type.

Ключевые столбцы, такие как order_id, order_date и revenue, не должны содержать нулевых значений. Для каждого из них проверьте df['col'].notna().all(). Чтобы сообщение проверки помогало устранить проблему, укажите в нём количество нулевых значений: так оператор конвейера поймёт масштаб проблемы, а не только узнает о том, что проверка завершилась неудачно.

Для числовых столбцов часто существуют допустимые диапазоны значений. Выручка не может быть отрицательной. Количество должно быть положительным целым числом. Цена за единицу должна быть больше нуля. Явно проверяйте эти ограничения: строка с отрицательной выручкой, которая пройдёт проверку, уменьшит итоговые суммы во всех последующих агрегированиях, причём никаких ошибок не возникнет. Проверки диапазонов на раннем этапе обнаруживают ошибки ввода данных и сбои в вышестоящих системах.

После удаления дубликатов значения order_id должны быть уникальными. Проверьте df['order_id'].is_unique, чтобы убедиться, что это инвариант сохраняется при каждом запуске конвейера. Нарушение уникальности после удаления дубликатов указывает на ошибку в логике дедупликации или на новый источник данных, который не был очищен перед объединением с основным набором данных.

Для столбцов с конечным набором допустимых значений, например region или category, проверьте, что каждое значение входит в разрешённый набор. Это обнаруживает неожиданные значения, появившиеся после переноса системы или изменения процесса ввода данных в вышестоящей системе. Определите допустимые наборы в конфигурации конвейера, чтобы их было легко обновить при расширении бизнеса на новые регионы.

Проверьте, что все даты находятся в ожидаемом периоде для этого набора данных. Заказ с датой в будущем невозможен; заказ с датой до основания компании указывает на повреждённую запись. Определите MIN_DATE и MAX_DATE в конфигурации и проверьте, что столбец находится в заданных границах. Это также обнаруживает даты начала эпохи Unix (1970-01-01), возникшие из-за неправильного преобразования временных меток.

Резкое изменение количества строк по сравнению с предыдущим запуском конвейера — надёжный признак проблемы в вышестоящей системе. Сохраните количество строк предыдущего запуска в файле конфигурации и проверьте, что новое значение находится в допустимом диапазоне, например в пределах ±20 % от исторического значения. Если между запусками набор данных потерял 50 % строк, это почти наверняка означает, что экспорт данных неисправен.

Соберите все проверки в одну функцию validate(df), которую можно вызывать в начале каждого этапа конвейера. Если проверка не пройдена, она вызывает AssertionError с понятным описанием проблемы. Такой подход делает конвейер самодокументируемым: функция проверки представляет собой машиночитаемый контракт схемы для DataFrame на данном этапе.

В рабочих конвейерах жёсткое завершение из-за assert допустимо при разработке, но нежелательно в запланированной задаче. Замените отдельные проверки блоками try/except AssertionError, которые записывают сообщение об ошибке и при необходимости отправляют уведомление перед завершением работы. Благодаря этому система мониторинга фиксирует причину сбоя, а не только трассировку необработанного исключения.

В этом уроке Вы научились: проверять обязательные столбцы, типы данных, ограничения на нулевые значения и допустимость диапазонов, проверять уникальные ключи, категориальные значения и диапазоны дат, а также объединять все проверки в повторно используемую функцию validate() с ведением журнала. Далее мы рассмотрим пользовательские агрегирования с помощью apply() для столбцов и строк.