Writing Robust Tests for Data & Machine Learning Pipelines

eugeneyan

Начать здесь Статьи Выступления Прототипирование Обо мне

Как писать устойчивые тесты для пайплайнов данных и машинного обучения

[ engineering machinelearning production 🩷 ] · 19 мин чтения

В последнее время я размышлял о том, как тестируются пайплайны данных и машинного обучения, и особенно — почему одни тесты ломаются чаще других. И дело не в том, что новый код неверен; часто новый код корректен, но тесты всё равно падают и их приходится обновлять.

Здесь я разбираюсь, почему определённые тесты ломаются — некорректно — чаще остальных, и пытаюсь найти более удачный подход к тестированию пайплайнов. Мы начнём с простого пайплайна и протестируем его с помощью юнит-, схема- и интеграционных тестов. Затем я введу новые данные и логику, посмотрю, как ломаются тесты, и выведу из этого закономерности. В конце я предложу, как сделать тестирование пайплайнов менее хрупким.

Обзор уровней тестирования: юнит, интеграционный, функциональный и др. Пример пайплайна: поведенческие логи -> результат батч-инференса Пишем тесты для пайплайна: юнит, схема, интеграционные Добавляем новые данные (видимые импрессии) и логику в пайплайн Аддитивный vs. ретроактивный эффект новых данных и логики на тесты Смежные понятия: валидность, гранулярность и охват тестов Рекомендации, как сделать тесты пайплайнов устойчивее

Меньше охват теста = короче петля обратной связи

Прежде чем нырнуть в детали, краткий обзор разных уровней тестирования (в статье Atlassian перечислены аж семь). Если мы строим пайплайн рекомендательной системы, вот как уровни ложатся на конкретные тесты:

Unit: тестируем отдельные методы и классы. Integration: тестируем точки интеграции пайплайна — например, между обработкой данных и тренировкой модели, или проверяем, что обученные модели корректно деплоятся. Functional: проверяем выполнение бизнес-требований (например, исключение эротики, сворачивание фильмов одной серии, обновление рекомендаций по кликам в сессии и т.п.). End-to-end: проверяем, что вывод пайплайна/эндпоинт корректно интегрируется с сайтом и рекомендации отдаются правильно по user ID или поведению. Acceptance: визуальная инспекция и QA живой выдачи рекомендаций. Performance: нагрузочное тестирование, оценка задержек и пропускной способности. Smoke: вручную сформированные URL для запуска рекомендаций по обычно проверяемым QA элементам. A/B: оценка эффекта изменений пайплайна или модели.

Охват тестов растёт сверху вниз по списку. Чем шире охват, тем длиннее петля обратной связи: юнит-тест выполняется за миллисекунды, а A/B-тест требует времени на внедрение изменений плюс срок самого теста.

Здесь мы сосредоточимся на тестах с самой короткой петлёй обратной связи: юнит-тестах и интеграционных тестах. Конкретно — на тех, что можно запустить офлайн, локально или на лёгком облачном инстансе. (Хотя интеграционные тесты могут выполняться и в CI-окружении с обращением к другим сервисам или БД, такие мы исключаем.)

Также, хотя у пайплайнов данных и ML обычно есть дополнительные тесты на качество данных (например, Great Expectations, Deequ), оценку модели (например, метрики scikit-learn) и проверки поведения модели (например, CheckList, RecList), они обычно выполняются в проде с каждой новой партией данных или обновлением модели. Поэтому здесь мы их обсуждать не будем.

Пример пайплайна: поведенческие логи -> результат инференса

На верхнем уровне наш пример пайплайна (i) потребляет поведенческие логи на уровне запросов, (ii) преобразует их в события на уровне элементов, (iii) обучает наивную модель, оценивающую CTR (click-through rate), и (iv) выполняет батч-инференс по списку элементов.

Вот как выглядят поведенческие логи; это также пример данных, которые мы будем использовать в тестах. Они фиксируют каждый запрос на выдачу рекомендаций и элементы, которые были показаны (impress) или кликнуты. (В сыром виде такие логи обычно неструктурированы, но в этом пайплайне они уже распарсены и сохранены в удобном табличном виде.)

+--------------+--------------------------+--------------+--------------+ | request_id | impressions | event_item | event_type | |--------------+--------------------------+--------------+--------------| | r1 | ['i1', 'i2', 'i3', 'i4'] | | impress | | r2 | ['i2', 'i3', 'i4', 'i1'] | i2 | click | | r3 | ['i3', 'i4', 'i1', 'i2'] | | impress | | r4 | ['i4', 'i1', 'i2', 'i3'] | i3 | click | | r5 | ['i1', 'i2', 'i3', 'i4'] | | impress | +--------------+--------------------------+--------------+--------------+

Краткое описание каждой колонки:

request_id: ID запроса на выдачу рекомендаций. impressions: ID показанных элементов. event_item: ID кликнутого элемента; пусто, если клика не было. event_type: тип события (impress или click).

Для запроса r1 были показаны четыре элемента (i1, i2, i3, i4), но ни один не был кликнут. В r2 были показаны четыре элемента (i2, i3, i4, i1), и i2 — элемент на первой позиции — был кликнут.

Чтобы было удобнее оценивать CTR на уровне элемента, мы хотим преобразовать поведенческие логи уровня запроса в события уровня элемента, где каждое показывание и каждый клик — отдельная строка. Дополнительно нам нужна позиция элемента в момент показа или клика.

Вот как выглядит таблица событий для r2 после преобразования:

+--------------+--------+------------+--------------+ | request_id | item | position | event_type | |--------------+--------+------------+--------------| | r2 | i2 | 1 | impress | | r2 | i3 | 2 | impress | | r2 | i4 | 3 | impress | | r2 | i1 | 4 | impress | | r2 | i2 | 1 | click | +--------------+--------+------------+--------------+

Далее мы обучаем наивную модель, которая выучивает исторический CTR и оценивает CTR на уровне элемента. Эту модель затем используем для батч-инференса по списку ID элементов.

Вот как выглядит результат батч-инференса. Обратите внимание, что ожидаемый CTR для i5 равен -1. Так получается, потому что поведенческие логи у нас есть только для i1—i4, а i5 — новый или «холодный» элемент. Поскольку модель не может оценить CTR для i5, она возвращает заранее заданное null-значение (-1).

+-----------+----------------+ | item_id | expected_ctr | |-----------+----------------| | i1 | 0 | | i2 | 0.2 | | i3 | 0.2 | | i4 | 0 | | i5 | -1 | +-----------+----------------+

До сих пор я делился преимущественно примерами данных на разных этапах пайплайна. Если вам удобнее разбираться по коду — посмотрите pipeline.py здесь. Учтите, что код намеренно упрощён — его реализация не ключевая в этой статье.

Тесты реализации: юнит, схема, интеграционные

Для тестирования пайплайна добавим (i) построчные юнит-тесты, (ii) поколоночные юнит-тесты, (iii) потабличные юнит-тесты, (iv) тесты схемы и (v) интеграционные тесты. Используем базовую функциональность Pandas (assert_frame_equal) и PyTest (fixtures). Тестовые фикстуры — на GitHub.

Начнём с самой мелкой формы — построчных юнит-тестов, где мы тестируем метод, принимающий вход (строку) и возвращающий одно значение. Протестируем get_click_position, который принимает массив импрессий и кликнутый элемент и возвращает позицию кликнутого элемента. Тест писать легко, и его можно параметризовать различными входами и ожидаемыми выходами.

def get_click_position(impressions: List[str], click: str) -> int: """Returns the position of the clicked item based on the array of impressions.""" try: return impressions.index(click) + 1 except ValueError: return -1 # Unit test: Row level @pytest.mark.parametrize('impressions,click,expected', [(['i1', 'i2', 'i3', 'i4'], 'i1', 1), (['i1', 'i2', 'i3', 'i4'], 'i3', 3), (['i1', 'i2', 'i3', 'i4'], None, -1), (['i1', 'i2', 'i3', 'i4'], 'NA', -1)]) def test_get_click_position(impressions, click, expected): assert get_click_position(impressions, click) == expected

Дальше добавим поколоночные юнит-тесты для методов, которые принимают колонку или таблицу и возвращают колонку. Метод get_impress_positions принимает таблицу с ID запросов и событиями показов и возвращает колонку с позицией каждого показа. Поскольку ожидаемый выход — колонка, а не одно значение, тест становится громоздким, и задавать ожидаемые значения утомительно.

def get_impress_positions(df: pd.DataFrame) -> pd.Series: """Returns a column of impression positions. Note: This method assumes that impressions are sorted in ascending order of their position. """ positions = df.groupby('request_id').cumcount() + 1 positions = positions.reset_index(drop=True) return positions # Unit test: Column level def test_get_impress_position(logs): impress_events = logs.explode('impressions') impress_positions = get_impress_positions(impress_events) # Since each impress log has 4 items, expect impression positions to be five sets of 1-4 pd.testing.assert_series_equal(impress_positions, pd.Series([1, 2, 3, 4] * 5))

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

+--------+---------+-----------+ | item | click | impress | |--------+---------+-----------| | i1 | 0 | 5 | | i2 | 1 | 5 | | i3 | 1 | 5 | | i4 | 0 | 5 | +--------+---------+-----------+

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

def aggregate_events(events: pd.DataFrame) -> pd.DataFrame: """Returns a table of items and their aggregated impressions and clicks.""" events_agg = events.pivot_table(index=['item'], columns=['event_type'], values=['request_id'], aggfunc='count', fill_value=0) events_agg.columns = events_agg.columns.droplevel() events_agg = events_agg.reset_index() # Clear index name events_agg = events_agg.rename_axis(None, axis=1) return events_agg # Unit test: Table level def test_aggregate_events(events): result = aggregate_events(events) arr = [['i1', 0, 5], ['i2', 1, 5], ['i3', 1, 5], ['i4', 0, 5]] cols = ['item', 'click', 'impress'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected)

Также добавим тесты схемы в ключевых точках пайплайна. Здесь мы проверяем ожидаемые имена колонок и типы данных после преобразования логов уровня запроса в события уровня элемента. Этот тест проверяет минимальный набор колонок: пайплайн можно обновлять без падения теста, пока минимальный набор колонок присутствует с правильными типами данных.

# Schema check: Columns and datatypes def test_events_schema(events): expected_cols = {'request_id': np.object, 'item': np.object, 'position': np.int64, 'event_type': np.object} # Check all expected columns are present assert len(set(expected_cols.keys()) - set(events.columns)) == 0, \ f'{set(expected_cols.keys()) - set(events.columns)} columns missing!' # Check all column data types are correct for col, dtype in expected_cols.items(): assert events.dtypes[col] == dtype, \ f'Expected column {col} to be of {dtype} type but found {events.dtypes[col]} type!'

Наконец, добавим интеграционные тесты, которые берут фиксированный пример входа, прогоняют его через пайплайн и сравнивают результат с ожидаемым выходом. Ниже мы проверяем, что поведенческие логи корректно преобразуются в события.

# Integration test: Input logs to aggregated events def test_feature_pipeline(logs): impress_logs = logs click_logs = logs[logs['event_type'] == 'click'] impress_events = get_impress_item_and_pos(impress_logs) click_events = get_click_item_and_pos(click_logs) sample_events = pd.concat([impress_events, click_events]) result = aggregate_events(sample_events) arr = [['i1', 0, 5], ['i2', 1, 5], ['i3', 1, 5], ['i4', 0, 5]] cols = ['item', 'click', 'impress'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected)

Также проверим, что наша модель действительно обучается и корректно возвращает ожидаемые оценки CTR.

# Integration test: Aggregated events to batch inference def test_model_pipeline(events, model, items): model = model.fit(events) result = model.batch_predict(items) arr = [['i1', 0.0], ['i2', 0.2], ['i3', 0.2], ['i4', 0.0], ['i5', -1.0]] cols = ['item_id', 'expected_ctr'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected)

Интеграционные тесты, пожалуй, самые трудоёмкие — они опираются на корректную работу нескольких преобразований данных и шагов обучения. Поэтому они становятся неуправляемыми, если пайплайн часто обновляется и они так же часто падают. Тем не менее они полезны для подтверждения того, что пайплайн работает как задумано. Их ожидаемый выход также помогает объяснить, как пайплайн должен работать.

Вот результат запуска наших тестов.

> pytest -v tests/test_pipeline.py --cov=src --cov-report=term-missing Name Stmts Miss Cover Missing ----------------------------------------------- src/__init__.py 0 0 100% src/pipeline.py 42 0 100% ----------------------------------------------- TOTAL 42 0 100%

Ремарка: в этом простом примере код пайплайна умещается в один скрипт. В промышленных условиях пайплайн может распределяться по нескольким репозиториям и инфраструктуре — Spark (Scala, SQL), SageMaker (Python, Docker). Настроить и запустить офлайн-тесты по нескольким репозиториям возможно, но в зависимости от приложения и масштаба это может не стоить усилий.

Добавление новых данных или логики = обновление кода пайплайна

В текущем пайплайне логируемые импрессии — это серверные импрессии: все элементы в выдаче рекомендаций логируются как показанные пользователю, даже если он фактически их не видел.

Например, на Netflix только первые шесть элементов в выдаче рекомендаций видны на экране; чтобы увидеть остальные, нужно прокрутить вбок. С серверными импрессиями, даже если я не прокручивал, все элементы выдачи будут залогированы как показанные. Это приводит к расхождению между тем, что мы считаем увиденным пользователем, и тем, что он реально увидел, особенно на маленьких экранах вроде смартфонов.

Только небольшая часть элементов выдачи видна, если пользователь не прокручивает вбок.

Чтобы это исправить, добавим в пайплайн дополнительные данные: клиентские импрессии. Эти данные фиксируют, какие элементы реально были видны на экране. Тогда, если я не прокручивал выдачу вбок, только первые шесть элементов будут залогированы как клиентские импрессии.

Вот как выглядят обновлённые поведенческие логи. У нас появилась новая колонка для видимых импрессий. (Но клиентские данные несовершенны, поэтому у r3 там null.)

+--------------+--------------------------+--------------------------+--------------+--------------+ | request_id | impressions | impressions_visible | event_item | event_type | |--------------+--------------------------+--------------------------+--------------+--------------| | r1 | ['i1', 'i2', 'i3', 'i4'] | ['i1', 'i2', 'i3'] | | impress | | r2 | ['i2', 'i3', 'i4', 'i1'] | ['i2', 'i3'] | i2 | click | | r3 | ['i3', 'i4', 'i1', 'i2'] | | | impress | | r4 | ['i4', 'i1', 'i2', 'i3'] | ['i4', 'i1', 'i2', 'i3'] | i3 | click | | r5 | ['i1', 'i2', 'i3', 'i4'] | ['i1', 'i2'] | | impress | +--------------+--------------------------+--------------------------+--------------+--------------+

Нам нужно обновить пайплайн так, чтобы он учитывал только видимые импрессии по простой логике. Добавим методы для обновления импрессий (на уровне строки) перед обновлением колонки импрессий (на уровне таблицы).

def get_updated_impressions(impressions: List[str], impressions_visible: List[str]) -> List[str]: """Returns an array of updated impressions based on logs of visible impressions.""" if isinstance(impressions_visible, list) and len(impressions) >= len(impressions_visible): return impressions_visible return impressions def update_impression_col(df: pd.DataFrame) -> pd.DataFrame: """Returns a table of logs where impressions is updated to consider visible impressions.""" _df = df.copy() _df['impressions'] = _df.apply(lambda x: get_updated_impressions(x['impressions'], x['impressions_visible']), axis=1) _df = _df.drop(columns=['impressions_visible']) return _df

Когда мы пытаемся снова прогнать наши существующие тесты с новым входом и кодом пайплайна, без учёта параметризации 4 из 6 тестов падают из-за новых данных и логики.

> pytest -v tests/test_pipeline.py --cov=src --cov-report=term-missing tests/test_pipeline.py::test_get_click_position[impressions0-i1-1] PASSED [ 11%] tests/test_pipeline.py::test_get_click_position[impressions1-i3-3] PASSED [ 22%] tests/test_pipeline.py::test_get_click_position[impressions2-None--1] PASSED [ 33%] tests/test_pipeline.py::test_get_click_position[impressions3-NA--1] PASSED [ 44%] tests/test_pipeline.py::test_get_impress_position FAILED [ 55%] tests/test_pipeline.py::test_aggregate_events FAILED [ 66%] tests/test_pipeline.py::test_events_schema PASSED [ 77%] tests/test_pipeline.py::test_feature_pipeline FAILED [ 88%] tests/test_pipeline.py::test_model_pipeline FAILED [100%]

Добавление новых данных/логики: аддитивный vs. ретроактивный эффект

Поскольку пайплайн обновился, нам нужно добавить тесты для метода get_updated_impressions — его легко параметризовать.

def get_updated_impressions(impressions: List[str], impressions_visible: List[str]) -> List[str]: """Returns an array of updated impressions based on logs of visible impressions.""" if isinstance(impressions_visible, list) and len(impressions) >= len(impressions_visible): return impressions_visible return impressions # Unit test: Row level (added) @pytest.mark.parametrize('impressions,impressions_visible,updated_impressions', [(['i1', 'i2', 'i3', 'i4'], ['i1', 'i2', 'i3'], ['i1', 'i2', 'i3']), (['i1', 'i2'], ['i1', 'i2', 'i3'], ['i1', 'i2']), (['i1', 'i2', 'i3', 'i4'], None, ['i1', 'i2', 'i3', 'i4'])]) def test_get_updated_impressions(impressions, impressions_visible, updated_impressions): assert get_updated_impressions(impressions, impressions_visible) == updated_impressions

А что с существующими тестами?

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

Существующие поколоночные юнит-тесты нужно обновить, потому что теперь мы исключаем невидимые импрессии. Новый ожидаемый выход приходится составлять вручную.

# BEFORE - Unit test: Column level def test_get_impress_position(logs): impress_events = logs.explode('impressions') impress_positions = get_impress_positions(impress_events) # Since each impress log has 4 items, expect impression positions to be five sets of 1-4 pd.testing.assert_series_equal(impress_positions, pd.Series([1, 2, 3, 4] * 5)) # AFTER - Unit test: Column level (updated) def test_get_impress_position(logs): logs = update_impression_col(logs) impress_events = logs.explode('impressions') impress_positions = get_impress_positions(impress_events) # Update impress positions (too brittle?) pd.testing.assert_series_equal(impress_positions, pd.Series([1, 2, 3, 1, 2, 1, 2, 3, 4, 1, 2, 3, 4, 1, 2]))

Существующие потабличные юнит-тесты нужно обновить, потому что отфильтрованные импрессии меняют результаты агрегации.

# BEFORE - Unit test: Table level def test_aggregate_events(events): result = aggregate_events(events) arr = [['i1', 0, 5], ['i2', 1, 5], ['i3', 1, 5], ['i4', 0, 5]] cols = ['item', 'click', 'impress'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected) # AFTER - Unit test: Table level (updated) def test_aggregate_events(events): result = aggregate_events(events) arr = [['i1', 0, 4], # Updated impress count ['i2', 1, 5], ['i3', 1, 4], # Updated impress count ['i4', 0, 2]] # Updated impress count cols = ['item', 'click', 'impress'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected)

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

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

# BEFORE - Integration test: Aggregated events to batch inference def test_model_pipeline(events, model, items): model = model.fit(events) result = model.batch_predict(items) arr = [['i1', 0.0], ['i2', 0.2], ['i3', 0.2], ['i4', 0.0], ['i5', -1.0]] cols = ['item_id', 'expected_ctr'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected) # AFTER - Integration test: Aggregated events to batch inference (updated) def test_model_pipeline(events, model, items): model = model.fit(events) result = model.batch_predict(items) arr = [['i1', 0.0], ['i2', 0.2], ['i3', 0.25], # Update expected CTR ['i4', 0.0], ['i5', -1.0]] cols = ['item_id', 'expected_ctr'] expected = pd.DataFrame(arr, columns=cols) pd.testing.assert_frame_equal(result, expected)

Подведём итог:

Юнит-тесты (уровень строки): существующие тесты не меняются; добавляем тесты для новых методов. Юнит-тесты (уровень колонки): обновляем существующие тесты. Юнит-тесты (уровень таблицы): обновляем существующие тесты. Тест схемы: существующие тесты не меняются; опционально добавляем тесты для новых колонок. Интеграционные тесты: обновляем существующие тесты.

На этом простом примере видно, что эффект новых данных или логики (i) аддитивен для построчных юнит-тестов и тестов схемы, но (ii) ретроактивен для поколоночных/потабличных юнит-тестов и интеграционных тестов. В первом случае мы просто добавляем новые тесты, не трогая существующие. Во втором — нужно обновлять большинство существующих тестов: они хрупки к изменениям данных и логики пайплайна.

Также, учитывая, что вторая группа тестов требует больше времени на написание и обновление, становится понятно, почему мы тратим на них больше времени, чем на собственно код фич.

В пайплайне и тестах выше обновление тестов не было слишком обременительным — приходилось лишь править код, генерирующий вход и ожидаемый выход. К тому же таблицы были маленькими, с небольшим числом строк и колонок. Однако если данных больше и они в менее доступном из IDE формате (например, Parquet), обновлять входные и ожидаемые выходные данные по множеству строк и колонок будет мучительно.

Один из способов с этим справиться — генерация данных и property-based testing. Генерация данных означает создание фейковых данных по спецификациям: числовым распределениям и допустимым категориальным значениям. Помогают такие библиотеки, как Faker и Hypothesis.

Чтобы применить property-based testing к нашему пайплайну, нужно (i) сгенерировать много данных, (ii) прогнать их через пайплайн и (iii) проверить, что выход удовлетворяет заранее определённым свойствам. (В отличие от этого, всё, что мы обсуждали выше, — это example-based testing: даны вход и фиксированный выход, мы проверяем равенство.) Среди библиотек, поддерживающих property-based testing: Hypothesis, Pandera (через Hypothesis и с дополнительными проверками схемы), Giskard (пертурбации текстовых, графических и табличных входов и тесты на дисперсию модели).

Property-based testing выглядит многообещающе. Я какое-то время использовал для него Faker. Тем не менее у меня есть сомнения в его способности тестировать преобразования данных на уровне колонок и таблиц. К тому же property-based testing fuzzy: мы подаём в код фаззированные сгенерированные данные и смотрим, упадёт ли он. Это помогает массово ловить краевые случаи, но я не уверен, что его легко применить для тестирования конкретной логики — например, проверки правильного ранжирования элементов до и после бизнес-логики (диверсификации, бустинга).

Смежные понятия: валидность, гранулярность и охват тестов

Если тест ломается и его приходится часто обновлять — это валидный тест? Возможно, нет. Тест должен защищать от регрессий: если его приходится постоянно менять, скорее всего, он работает некорректно.

Подобрать правильную гранулярность теста бывает непросто, но именно она делает тестирование более устойчивым. Проверяя точные значения в интеграционных тестах выше, я слишком тесно связал ожидаемый выход с деталями реализации. Вероятно, стоило проверять более крупные характеристики — количество строк, колонок, уникальных ID элементов. Это сделало бы те интеграционные тесты устойчивее и они не ломались бы при добавлении новых данных.

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

Много построчных и схема-тестов, по чуть-чуть всего остального

Резюмируя: построчные юнит-тесты и тесты схемы требуют меньше усилий на написание и поддержку. Они также устойчивы к новым данным и бизнес-логике. К тому же исчерпывающие построчные тесты снижают зависимость от тестов более высокого уровня. Я применяю их щедро.

А вот поколоночные/потабличные юнит-тесты и интеграционные тесты обычно хрупкие. Подобрать правильную гранулярность для них непросто, но именно это делает тестирование устойчивее. Несмотря на минусы, такие тесты дают незаменимую страховку, хотя ими стоит пользоваться экономно, иначе они станут слишком обременительными в поддержке. Я привык полагаться на даваемую ими уверенность, но в будущем, вероятно, буду использовать их меньше.

А что вы думаете про тестирование пайплайнов данных и машинного обучения? Какие подходы оказались эффективными у вас? Очень хотел бы услышать ваши мысли!

Дальнейшее чтение

The Different Types of Software Testing What are the Differences between Unit Tests, Integration Tests, Smoke Tests, etc? Effective Testing for Spark Programs The Mechanics of Testing Large Data Pipelines Test Strategies for Data Processing Pipelines Engineering Data Quality Testing Features with PyTest Intro to Property-based Testing in Python What is Property Based Testing? Testing in Microsoft Recommenders

Спасибо David Said, Jim Dowling, Shreya Shankar, Mike Lam и Lars Albertsson за обсуждения по теме. OG-картинка — Sam Loyd на Unsplash.

Если эта статья вам пригодилась, цитируйте её так:

Yan, Ziyou. (Sep 2022). Writing Robust Tests for Data & Machine Learning Pipelines. eugeneyan.com. https://eugeneyan.com/writing/testing-pipelines/.

или

@article{yan2022testing, title = {Writing Robust Tests for Data & Machine Learning Pipelines}, author = {Yan, Ziyou}, journal = {eugeneyan.com}, year = {2022}, month = {Sep}, url = {https://eugeneyan.com/writing/testing-pipelines/} }

К рассылке присоединились 11 800+ читателей, получающих обновления о машинном обучении, RecSys, LLM и инженерии.