Проверка свежести данных с помощью GX

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

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

Great Expectations (GX) можно использовать в целях мониторинга и проверки свежести Ваших корпоративных данных. Используя функции, предлагаемые GX Cloud и GX Core, Вы сможете устанавливать проверки, обеспечивающие поддержание желаемого уровня свежести Ваших корпоративных данных. В этой статье мы поговорим о том, как эффективно управлять и проверять свежесть данных с помощью GX, гарантируя высокое качество и актуальность используемой Вами информации.

Необходимые условия

Эта статья предполагает, что Вы уже знакомы с основными компонентами и рабочими процессами GX.

Предварительный обзор данных

reading_id	sensor_id	tempera ture_k	reading_ts	created_at
1001	factory-1	305.00	2024-11-22 13:38:07	2024-11-22 13:47:00
1002	factory-2	310.00	2024-11-22 13:38:54	2024-11-22 13:47:00
1003	factory-3	308.00	2024-11-22 13:39:54	2024-11-22 13:47:00
1004	factory-1	303.75	2024-11-22 13:40:10	2024-11-22 13:47:00
1005	factory-2	311.00	2024-11-22 13:40:53	2024-11-22 13:47:00

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

reading_ts фиксирует временную метку отражающую информацию о том, когда датчик снял показания;
created_at - временная метка создания строки в базе данных, указывающая на то, когда данные стали доступны для анализа

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

Ключевые ожидания по свежести данных

GX Cloud и GX Core предоставляют несколько вариантов использования ожиданий для проверки свежести информации:

Встроенные ожидания, рассмотренные в этой статье можно использовать по отношению к столбцам временных меток для создания проверок свежести. Встроенные ожидания можно добавлять через пользовательский интерфейс GX Cloud или в процессе GX Core Python;
Пользовательские ожидания SQL можно использовать как в GX Cloud, так и в GX Core (для проверки свежести данных на основе логики SQL);
В GX Core можно настроить классы для создания ожиданий, которые используют логику Python для определения и проверки свежести данных.

Expect column maximum to be between

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

Пример: Убедитесь в том, что самая последняя временная метка образовалась после указанного времени.

gxe.ExpectColumnMaxToBeBetween(
column="reading_ts",
min_value="2024-11-22 14:42:00",
)

Expect column minimum to be between

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

Пример: Убедитесь в том, что самые старые данные не старше определенной даты.

gxe.ExpectColumnMinToBeBetween(

     column="reading_ts",
     min_value="2024-11-22 00:00:00",
 )

Подсказки GX

Используйте ExpectColumnMaxToBeBetween для проверки самой последней точки данных, чтобы убедиться в том, что Ваш набор данных все еще актуален. Используйте ExpectColumnMinToBeBetween для проверки самой старой точки данных, что полезно для проверки политики хранения данных или выявления старых записей. Комбинированное использование обеих функций позволяет обеспечить комплексную проверку свежести информации.

Примеры

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

Примеры, приведенные в этой статье показывают, как использовать функции GX Cloud и GX Core для создания и запуска ожиданий, учитывающих динамическую природу проверок свежести, включая использование функций типа now()и разницы временных меток.

Создайте пользовательское SQL-выражение свежести с помощью GX Cloud

Цель: создание пользовательских SQL-ожиданий для проверки свежести данных и организация ежечасного выполнения проверки данных.

Используйте пользовательский интерфейс GX Cloud для того, чтобы выполнить следующие шаги:

Используя следующую строку подключения для создания источника данных Postgres, Вы сможете создать актив данных для таблицы freshness_sensor_readings:

postgresql+psycopg2://try_gx:try_gx@postgres.wo

Используя запрос, приведенный ниже, создайте пользовательское SQL-ожидание для актива данных freshness_sensor_readings, которое показывает, что данные датчика будут доступны в базе данных не более чем через 10 минут после их первоначального захвата на датчике.

select *
 from {batch}
 where extract(epoch from (age(created_at, reading_ts))) > 10*60

Добавьте второе пользовательское SQL-ожидание на тот же актив данных, который показывает, что новые показания датчиков поступили в базу данных в течение последних 5 минут.

select *
 from (
    select max(created_at) as most_recent_reading
    from {batch}
 ) t
 where extract(epoch from (age(current_timestamp, most_recent_reading))) > 5*60

Отредактируйте активное расписание проверки, чтобы изменить частоту повторяющихся проверок данных. Выберите Частота каждые 1 час, чтобы выполнять повторяющиеся проверки свежести каждый час.
Просмотрите результаты проверки на активе данных freshness_sensor_readings после выполнения проверки.

Результат: Одно ожидание свежести проходит, а другое - нет.

Ожидание свежести, что показания датчиков доступны в базе данных не более чем через 10 минут после первоначального захвата, проходит, поскольку это условие справедливо для всех показаний датчиков, накопленных в базе данных.
Ожидание свежести, согласно которому новые показания датчиков поступили в базу данных в течение последних 5 минут, не выполняется, поскольку данные выборки представляют собой показания, полученные в прошлом. Если бы показания датчиков поступали регулярно, это ожидание могло бы проверить, поступают ли свежие данные в требуемый промежуток времени.

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

Создайте пользовательский класс ожиданий свежести с помощью GX Core

Цель: создание пользовательского класса Expectation в GX Core для проверки свежести данных относительно текущего момента времени.

Запустите следующий рабочий процесс GX Core:

import datetime
 import great_expectations as gx
 import great_expectations.expectations as gxe
 # Create Data Context.
 context = gx.get_context()
 # Connect to sample data and create Data Source, Data Asset, Batch Definition, and Batch.
 CONNECTION_STRING = "postgresql+psycopg2://try_gx:try_gx@postgres.workshops.greatexpectations.io/gx_learn_data_quality"
 data_source = context.data_sources.add_postgres(
     "postgres database", connection_string=CONNECTION_STRING
 )
 data_asset = data_source.add_table_asset(
     name="sensor readings", table_name="freshness_sensor_readings"
 )
 batch_definition = data_asset.add_batch_definition_whole_table("batch definition")
 batch = batch_definition.get_batch()
 # Define the custom Expectation class by subclassing the built-in ExpectColumnMaxToBeBetween Expectation.
 class ExpectSensorDataToBeFresh(gxe.ExpectColumnMaxToBeBetween):
     """Custom Expectation class to validate the freshness of sensor readings in the database."""
     column: str = "created_at"
     min_value: datetime.datetime = datetime.datetime.now() - datetime.timedelta(
         minutes=5
     )
     description: str = "New sensor readings should have arrived in the database within the last 5 minutes."
 # Validate the sample data with the custom freshness Expectation.
 validation_result = batch.validate(ExpectSensorDataToBeFresh())
 print(f"Freshness check passed: {validation_result['success']}")
 print(f"Most recent reading timestamp: {validation_result['result']['observed_value']}")

Результат:

Freshness check passed: False
 Most recent reading timestamp: 2024-11-22 14:49:00

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

Решение GX: GX Core позволяет использовать пользовательскую логику на основе Python для определения и проверки свежести данных. Пользовательские классы ожиданий в GX Core можно использовать для дополнения и расширения рабочих процессов GX Cloud.

Определение и установка пороговых значений свежести

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

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

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

С учетом всех этих факторов подходящими пороговыми значениями свежести для данных о розничных продажах могут быть:

В часы пик: последняя временная метка транзакции произошла в течение последних 30 минут.
В нерабочее время: последняя временная метка транзакции произошла в течение последних 2 часов.

Пороговые значения достаточно жесткие, они нужны для того, чтобы можно было быстро обнаружить устаревшие данные, которые могут негативно повлиять на бизнес. Пороговые значения можно установить с помощью параметра GX ExpectColumnMaxToBeBetween для столбца временной метки транзакции и запланировать его выполнение в GX Cloud каждый час.

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

Сферы применения

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

Аналитика данных СМИ

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

Решение GX: Применяя такие параметры, как ExpectColumnMaxToBeBetween и ExpectColumnMinToBeBetween, к столбцам временных меток в данных социальных сетей, агентство может убедиться в том, что оно фиксирует и анализирует самые последние взаимодействия и разговоры клиентов. GX позволяет агентству устанавливать соответствующие пороги свежести в зависимости от потребностей своих клиентов и динамичной природы социальных сетей.

Интеллектуальное производство и предиктивное обслуживание

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

Решение GX: GX позволяет проверять свежесть данных датчиков с помощью таких ожиданий, как ExpectColumnMaxToBeBetween и ExpectColumnMinToBeBetween. Настроив эти ожидания в своих конвейерах данных, компания сможет гарантировать, что данные точно отражают текущее состояние машин, что позволяет моделям предиктивного обслуживания выявлять потенциальные проблемы на уже на самых ранних стадиях и планировать профилактическое обслуживание заранее.

Автомобили и управление автопарком в режиме реального времени

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

Решение GX: GX позволяет установливать такие ожидания, как ExpectColumnMinToBeBetween и ExpectColumnMaxToBeBetween, на столбцы временных меток в данных об автомобиле. Включив эти ожидания в конвейер данных, компания может убедиться в том, что данные с подключенных автомобилей действительно актуальные, что позволяет отслеживать автопарк в режиме реального времени и быстро реагировать на любые отклонения от запланированных маршрутов.

Самые распространенные ошибки, допускаемые при проверке свежести данных

Очень часто команды по работе с данными сталкиваются со следующими ошибками проверки свежести данных:

Не определены четкие SLA: Команды должны устанавливать и документировать соглашения об уровне обслуживания (SLA), определяющие, насколько свежими должны быть те или иные наборы данных;
Не учтена актуальность пайплайна: Данные могут стать неактуальными абсолютно на любом этапе - от получения данных до их преобразования. Учитывайте общее отставание конвейера, а не только свежесть исходной системы. Добавьте проверки достоверности данных после прохождения ключевых этапов обработки;
Усреднение временных меток может скрыть проблемы: Вычисление средней временной метки может привести к тому, что общие данные будут казаться обманчиво свежими. Вместо этого используйте минимальные и максимальные значения, чтобы охватить весь диапазон;
Неправильная выборка данных для проведения проверки свежести: При проверке свежести проверяйте весь набор данных или используйте статистически обоснованный метод выборки.

Дальнейшие шаги

Придерживаясь лучших практик и понимая наиболее распространенные «подводные камни» проверки свежести, Вы сможете обеспечить актуальность данных в своем пайплайне. Внедрение проверки свежести с помощью GX позволит выявлять малейшие проблемы уже на самой ранней стадии, предотвращать потенциальные серьезные проблемы и поддерживать высокие стандарты качества данных.

Стремясь к поддержанию высокого качества данных, помните, что свежесть данных - это всего лишь один из важнейших аспектов комплексной стратегии контроля качества данных. Изучив всю нашу серию статей о качестве данных, Вы узнаете о том, как с помощью GX интегрировать другие важные аспекты качества данных, такие как целостность, полнота, распределение и схема.

← Предыдущая статья

Проверка распределения данных с помощью GX

Следующая статья →

Проверка целостности данных с помощью GX