Профилирование данных и повышение качества DAG-конвейеров Apache AirFlow с Whylogs

Как повысить качество данных и пакетных конвейеров с их обработки в Apache AirFlow с Python-библиотекой Whylogs. Что это за средство регистрации и профилирования, как оно работает, каким образом совместимо с DAG-графом задач Apache AirFlow и чем полезно дата-инженеру.

Что такое Whylogs и зачем это Apache AirFlow

Apache AirFlow активно используется дата-инженерами в качестве отказоустойчивой платформы управления пакетными конвейерами обработки больших данных, помогая создавать, планировать и контролировать процессы с помощью удобного API. Он гарантирует, что запланированные задания будут выполняться по заранее запланированному расписанию и сообщит, если отдельные задачи завершились неудачно. Однако, знания, что batch-процесс выполнился успешно и вовремя, недостаточно для надежности. На практике дата-инженеру нужна более детальная информация о ходе обработки данных с помощью Apache AirFlow. Для этого приходится использовать сторонние решения. Одним из них является Python-библиотека с открытым исходным кодом для регистрации любых данных Whylogs. Она позволяет пользователям проверять качество данных с помощью богатого API, включая запуск пакета проверки ограничений, создание отчетов о дрейфе данных и пр.

С Whylogs пользователи могут создавать сводки своих наборов данных, называемые профилями, чтобы использовать их для отслеживания изменений в датасете, создавать ограничения, т.е. эталонные допустимые диапазоны данных, а также быстро визуализировать ключевую сводную статистику. Эти три простые функции обеспечивают множество следующих вариантов использования для Data Analyst’ов, инженеров данных и специалистов по машинному обучению:

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

Whylogs можно запускать в Python-средах, а также Apache Spark, причем как PySpark, так и Scala, для различных типов данных. Также библиотека интегрирована с Pandas, AWS Sagemaker, MLflow, Flask, Ray, RAPIDS, Apache Kafka и другими фреймворками.

Чтобы максимально эффективно использовать логи Whylogs в конвейерах AirFlow, дата-инженеры ML-компании WhyLabs создали соответствующий провайдер — оператор AirFlow, который можно бесплатно установить через менеджер пакетов pip:

$ pip install airflow-provider-whylogs

Как этот пакет позволяет отслеживать качество данных в конвейерах Apache AirFlow, мы рассмотрим далее.

Как повысить качество данных с помощью средства их регистрации

Конвейер Airflow определяется в направленном ациклическом графе (DAG), что означает, что каждое задание будет иметь начальную задачу и зависимости, которые всегда будут выполняться в одном направлении без возврата к началу после завершения. Если по какой-либо причине предыдущая задача завершается сбоем, то нижестоящие не будут выполняться, что может привести к отказу всего производственного процесса или некорректным результатам. Например, успешно выполненное задание машинного обучения могло привести к смещению данных, что без надлежащего мониторинга будет обнаружено слишком поздно. С помощью журнала Whylogs мониторинг данных и машинное обучение можно выполнять непрерывно. А провайдер для AirFlow позволит расширить существующие конвейеры, чтобы сделать пакетные ETL-процессы более надежными.

После установки библиотеки Whylogs, можно ее сразу использовать. Для этого сперва нужно профилировать данные и сохранить туда, где приложение AirFlow сможет получить доступ к данным. Одним из возможных мест для хранения пользовательских профилей является локальная файловая система, если AirFlow запускается локально на экземпляре виртуальной машины, а ее части обработки и хранения являются общими. Также можно отправить Spark-задание в AWS EMR и сохранить профиль в корзине AWS S3.

Рассмотрим простой пример создания пользовательского профиля из датафрейма Python-библиотеки Pandas, который считан из CSV-файла:

import whylogs as why
import pandas as pd
df = pd.read_csv("path/to/file.csv")
results = why.log(df)

Профили являются ядром библиотеки Whylogs. Они фиксируют ключевые статистические свойства данных, такие как распределение, количество пропущенных значений и широкий спектр настраиваемых пользовательских показателей. Собирая эти сводные статистические данные, можно точно представлять данные и использовать все сценарии. Профили Whylogs обладают свойствами, которые делают их идеальными для регистрации данных: они эффективны, настраиваются и объединяются.

Профили детально описывают набор данных, который они представляют. Статистика, собираемая профилями Whylogs, легко настраивается, поскольку для разных типов данных и вариантов использования требуются разные метрики, а пользователям нужна возможность легко определять собственные трекеры для этих метрик. Именно настраиваемость журналов Whylog позволяет использовать текстовые, графические и другие сложные средства отслеживания данных. Наконец, одной из самых мощных функций профилей Whylogs является возможность их объединения для формирования новых профилей. Это позволяет вести журнал для распределенных и потоковых систем, а также дает пользователям возможность просматривать агрегированные данные с любой степенью детализации по времени.

Установка ограничений данных в ваших профилях позволяет получать уведомления, когда данные не соответствуют ожиданиям, т.е. вести модульное тестирование данных и некоторый базовый мониторинг. А встроенный визуализатор профилей наглядно отображает данные, чтобы Data Scientist мог убедиться, что ML-модели готовы к производству. Также на основе профилей Whylogs можно создать ограничения, чтобы быстро проверить, что данные выглядят так, как надо. Следующий пример кода показывает создание ограничений из ранее созданного пользовательского профиля:

from whylogs.core.constraints import Constraints, ConstraintsBuilder, MetricsSelector, MetricConstraint

profile_view = why.log(df).view()
builder = ConstraintsBuilder(profile_view)

builder.add_constraint(MetricConstraint(
    name="col_name >= 0",
    condition=lambda x: x.min >= 0,
    metric_selector=MetricsSelector(metric_name='distribution', column_name='col_name')
))
constraints: Constraints = builder.build()
constraints.report()

Больше полезных приемов администрирования и эксплуатации Apache AirFlow для дата-инженерии и аналитики больших данных вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:

• Data Pipeline на Apache Airflow
• Data pipeline на Apache AirFlow и Arenadata Hadoop

Источники

1. https://whylabs.ai/blog/posts/data-quality-monitoring-in-apache-airflow-with-whylogs
2. https://github.com/whylabs/whylogs
3. https://pypi.org/project/airflow-provider-whylogs/