Единственное, что есть в нашей жизни постоянного, — это изменения. (Цитата из книги "Конвоиры зари" Дона Уинслоу). Фраза чуть отредактирована, но не об этом ли пойдёт речь? Любые изменения касающиеся условий и семантики обмена данными должны явно отслеживаться. На практике зачастую это работает, вот так:

- Санечка нужно поменять тип колонки в таблице. Как думаешь это заденет кого-нибудь?

- Дядь меняй!, проверим как раз кто этой таблицей пользуется.

p.s. Cитуация вымышленная, любые совпадения случайны.

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

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

• Что такое дата контракт?

• Есть ли единый стандарт?

• Может уже используете дата контракты?

• Когда нужно использовать дата контракты?

• Как мы используем дата контракт?

Что такое дата контракт?

Data сontract - это соглашение между поставщиком и потребителем об условиях обмена данными. В качестве сторон взаимодействия выступают data owners, data analysts, data engineers, bi - analysts, data scientists, developers, services, erp, crm - системы и другие заинтересованные стороны. Имеется ввиду не только взаимодействие пользователей по какому-либо data-SLA (Service Level Agreement), но и прозрачная работа самих дата продуктов. Концепция встраивания контрактов в data pipeline обусловлена обеспечением целостности данных. Сам же дата контракт может существовать в виде описания всей необходимой информации на страничке в confluence, но чаще всего эта история не поддерживается. Много зависит от культуры работы с данными в компании.

Обычно спецификация дата контракта включает в себя:

• Общую информацию.

• Модель данных.

• Биллинговую информацию.

• Проверки качества данных.

• И другое опционально.

Информация необходима для соблюдения регламентов работы с данными. Способы реализации индивидуальны в определённых ситуациях. Йохен Крист и доктор Саймон Харрер представили свой вариант спецификации Data Сontract. Коллеги проделали огромную работу и реализовали инструмент для работы с контрактами Data Сontract CLI. Проект набирает обороты и начинает потихоньку закрывать эту нишу? Постараемся ответить на этот вопрос позже. Отметим, что на рынке это не единственный продукт по работе с контрактами. Чуть больше узнать можно по ссылочкам об инструменте, выделим лишь только пару важных аспектов:

• Есть версионирование дата контрактов

• Data quality - по спецификациям soda, montecarlo, great expectations и др.

• Генерация схемы для спецификации на базе существующих данных, например на основе ddl.

• Есть движки для postgres, kafka, s3, snowflake.

• Возможность работы со стандартом.

• Интеграция с Data Mesh Manager & OpenTelemetry

Помимо отслеживания изменений в data pipeline через контракт, хорошей практикой будет их хранение в дата каталогах для централизованной работы всех заинтересованных пользователей, а также применение мониторинга проверок в соответствии со спецификацией контракта по качеству данных. Всё это прекрасно и выглядит как решение проблемы несогласованности, регулирования взаимодействия по дата обмену. Да и уже существуют стандарты дата контрактов, они развиваются, вокруг этого всего складывается активное сообщество, но до сих об этом больше говорят, чем используют. Даже есть проекты книг по дата контрактам, наиболее известная - "Data Contracts - Developing Production Grade Pipelines at Scale by Chad Sanderson, Mark Freeman". Дата контракт становится похож на таблетку от всех болезней!

Есть ли единый стандарт?

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

Лучшее решение это использование совокупности практик. Но централизованное обеспечение совместимости трактов данных посредством управления единым стандартом то к чему нужно стремиться, например к Open Data Contract Standard или Data Product Descriptor Specification. Ссылочки на подобные ресурсы оставим ниже, чтобы вы нашли подходы для решения своих задач и поняли что для вас стандарт дата контракта.

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

Может уже используете дата контракты?

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

Валидация данных с помощью pydantic моделей, также может являться частью соблюдения регламентов контракта. Использование решений для построения data lineage неразделим с дата контрактами, поддержания каталога данных и тракта работы дата продуктов. Например, использование OpenLineage в связке с Airflow или комплексных решений OpenMetadata, DataHub, Airbyte для управления данными - позволяет прозрачнее вытроить дата процессы.
Поэтому выстраивание культуры работы с данными не важно какими инструментами, включает в себя каком-либо уровне уже неявную имплементацию дата контрактов.

Когда нужно использовать дата контракты?

Дата контракт не должен применяться как панацея. Ключевая проблема которую решает контракт - это возможность проактивно влиять на любые дата изменения. В компании, где правильно выстроены процессы data governance использование контракта может быть излишним. С другой стороны команда просто не достаточно зрелая и в них просто нет необходимости или соглашения на уровне договорённости могут быть достаточными.

Ответственность за любые изменения в данных, как правило, ложится на владельца данных совместно со специалистом по качеству данных, но за частую это один разработчик, создавший дата продукт, да и в мире активно превалирует микросервисная архитектура со своими особенностями. Она порождает data mesh и тут меняются подходы к работе с децентрализированномым хранением данных. В подобной парадигме отслеживание любых изменений становится затруднительным, а значит появляется необходимость этот кейс закрывать. Создание дата контракта во многом зависит от выбранного технологического стека, дата архитектуры, культуры работы с данными. В конечном счёте можно понаделать дата контрактов, но на деле их никто реализовывать не будет, так как выполнение самого контракта не так сложно как встраивание в pipeline. И бесспорно нужно задумываться о целесообразности их использования, например в работе с real-time данными имплементация контрактов можно повлиять на скорость загрузки данных.

Обеспечение согласованности обмена данными на уровне соглашений пользователей с помощью дата контрактов хорошо описан в книге Джеймса Денсмора - Data Pipelines Pocket Reference. Идея заключается в возможности создания автоматических уведомлений всех заинтересованных сторон при изменении контракта. В основе лежит встраивание хуков в CI/CD процессы для отслеживания состояния дата контрактов. Концепция ясна, но в действительности построение процесса, скорее всего будет сопровождаться сложностями с синхронизацией версий дата контрактов, разделением ролей по доступу и подобными штуками. В свою очередь Чад Сандерсон и Адриан Крейцигер в серии статей An Engineer's Guide to Data Contracts как раз описали процесс имплементации дата контрактов в data pipeline. Подход работает на отслеживании изменений в дата продукте, используя сравнения спецификаций контрактов на базе CDC и потоковой обработки. Тем самым предложили баланс между технической сложностью внедрения и прозрачностью дата процессов.

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

Как мы используем дата контракт?

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

Реализация заключается в создании дага, который будет запускаться с помощью rest api airflow после подготовки данных для горячего хранилища. Шаг сбора и подготовки данных из external source в слоях greenplum мы опустим так как нам он не так интересен, но он инициализирует запуск дага по завершении предрасчёта. Даг для переливки подготовленных витрин из источника в целевую бд состоит из 5 шагов:

• Проверка наличия контракта для переноса данных

• Валидация схемы данных источника и целевой таблицы

• Проверка качества данных в таблице источнике

• Обновление данных в целевой таблице

• Удаления неактуальных данных в целевой таблице

Последние четыре шага являются тасками в рамках выполнения дага. Каждая таска работает на базе KubernetesPodOperator c переданными параметрами и собранным докер образом. Питонячий код таски строиться по подходу DDD (Domain Driven Design). Все настройки для подключений хранятся в соединениях airflow, при конфигурации самого дага все они пробрасываются в поду. Первый шаг выполняется для каждой из четырёх предыдущих тасок. Это основной шаг, который формирует все конфиги для подключения к базам данных и настройки для работы с soda, а также проверяет наличие дата контракта на перенос данных.

Ключевой компонент репозитория proxy mapping data contract - он же мэтчинг контрактов для передачи данных. Представлен он в виде словаря, ключом которого является data_contract_id (например, "hot_storage1"), а значение модель ContractMapping, содержащая сущности для переноса данных по конкретному контракту:

from types import MappingProxyType

from dag.domain import (
    HotStorageSchema1,
)
from dag.domain.contract_map import ContractMapping
from dag.repos import (
    HOT_STORAGE_COLLECT1,
    HOT_STORAGE_DELETE1,
    HOT_STORAGE_INSERT1,
)

DATA_CONTRACTS = MappingProxyType(
    {
        'hot_storage1': ContractMapping(
            conn_target_name='target_db1',
            schema_validation=HotStorageSchema1,
            query_collect=HOT_STORAGE_COLLECT1,
            query_insert=HOT_STORAGE_INSERT1,
            query_delete=HOT_STORAGE_DELETE1,
        ),
    },
)

Описание сущностей ContractMapping:

from pydantic.main import BaseModel

class ContractMapping(BaseModel):
    conn_target_name: str
    schema_validation: type[BaseModel]
    query_collect: str
    query_insert: str
    query_delete: str

• conn_target_name - наименование подключения к целевой базе данных (должен соответствовать наименованию cоединения необходимой целевой бд в airflow)

• schema_validation - pydantic model для валидации переносимых данных

• query_collect - запрос для сбора данных из greenplum

• query_insert - запрос для вставки/обновления данных в горячем хранилище

• query_delete - запрос для удаления неактуальных данных

После подготовки образа с дата контрактом, создаётся конфиг для триггера дага. Он включает в себя помимо основных параметров ключ conf, который должен содержать:

• data_contract_id - наименование контракта соответствующего одному из наименований в DATA_CONTRACTS - proxy mapping data contract

• validation_checks - конфиг для формирования проверок качества данных (soda) в таблице источника , если нужно пропустить шаг validation_checks = "skip"

• target_table - наименование целевой таблицы

• target_schema - наименование целевой схемы

• gp_table - наименование таблицы источника

• gp_schema - наименование схемы источника

Даг запускается с конкретным конфигом для переноса данных из источника в целевое место с нужными проверками качества по конкретному data_contract_id. В процессе выполнения отслеживается:

• отсутствие запрашиваемого data_contract_id

• отсутствие одной из таблиц (источника, целевой)

• соответствующие контракту проверки качества данных в источнике

• атрибутный состав с типами используемых данных в источнике и целевой таблице.

• пропуск проверок качества данных

• валидация данных pydantic

Перенос данных в принципе может производить любым возможным способом - batches, fwd, через unlogged table, temp, copy или используя другие прелести бд. В текущем кейсе работает классический batch-upsert-трансфер. Любое падение дага и разбор инцидента проводится по data_contract_id в котором произошла ошибка. Proxy mapping data contract является некоторой общей точкой взаимодействия, объединяющая все дата контракты для миграции данных. Она позволяется проактивно реагировать на дата изменения не соответствующие регламенту контракта.

Заключение

Задели немного аспекты касающиеся темы data сontract. Пытались договориться о регламентах работы с данными и о самом использовании контрактов. Увидели большой пласт проделанной работы в этом направлении. Поблагодарили авторов. Посмотрели на контракты с другой стороны. И на этом не остановимся

Всем спасибо за уделённое время! Буду рад обсудить эту тему в комментариях! Всем добра!

Полезные ссылки:

1. Driving Data Quality with Data Contracts: A comprehensive guide to building reliable, trusted, and effective data platforms 1st Edition, Kindle Edition by Andrew Jones

2. Data Pipelines Pocket Reference by James Densmore

3. https://andrew-jones.com/

4. https://www.montecarlodata.com/

5. https://www.striim.com/

6. https://dataproducts.substack.com/

7. https://protobuf.dev/

8. https://datacontract.com/

9. https://blog.datahubproject.io/the-what-why-and-how-of-data-contracts-278aa7c5f294

10. https://github.com/agile-lab-dev/Data-Product-Specification

11. https://medium.com/exploring-the-frontier-of-data-products/emerging-everything-as-code-in-the-data-contract-standards-10ad7ad76fbb

12. https://github.com/bitol-io/open-data-contract-standard

13. https://dpds.opendatamesh.org/

14. https://opendataproducts.org/

15. https://github.com/paypal/data-contract-template

16. https://benn.substack.com/p/data-contracts

17. https://astrafy.io/the-hub/blog/technical/implementation-of-the-data-contracts-with-dbt-google-cloud-great-expectations-part-1

18. https://glossary.airbyte.com/term/data-contract/

19. https://www.decube.io/post/data-contracts-implementation-guide