AI-каталог данных с автоклассификацией и PII-детекцией

Data team тратит 15–30 минут на тегирование каждой таблицы, а с сотнями активов каталог устаревает в первый же месяц. Мы построили AI-каталог (LLM-каталог) с автотегированием таблиц и ML-классификацией метаданных — это data governance инструмент, который автоматически классифицирует датасеты, детектирует PII и строит lineage без ручного ввода. Наш сервис интегрируется за 2–4 недели без замены существующих storage-решений. Например, одна финтех-компания с 500 таблицами экономила 400 часов в месяц после внедрения, а точность тегирования выросла с 70% до 97%. Сравните: AI-каталог обрабатывает 500 активов за час, тогда как ручное тегирование заняло бы 125 часов — разница в 25 раз. Экономия на масштабе — до 1 млн руб. в год при 500 активах.

Почему AI-классификация точнее ручного тегирования?

Ручной каталог требует постоянного внимания: аналитики забывают описывать новые таблицы, а lineage строится постфактум. AI решает три ключевые задачи за секунды:

• Классификация: LLM (Claude 3.5 Sonnet, GPT-4o) генерирует описание, домен, теги и уровень чувствительности из DDL и семпла данных.
• Поиск: семантический поиск по описаниям и колонкам — находит связанные активы без точного совпадения.
• Lineage: автоматическое определение upstream/downstream связей через анализ SQL-запросов и кода.

Как мы строим AI-каталог?

Сканирование базы через SQLAlchemy, извлечение семпла, LLM-классификация — стек на Python с PyTorch для embeddings (all-MiniLM-L6-v2), ChromaDB для векторного поиска. Пример: сканируем PostgreSQL, получаем имена колонок и типы, отправляем в Anthropic API.

from anthropic import Anthropic
import sqlalchemy
import pandas as pd
import json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class DataAsset:
    asset_id: str
    name: str
    asset_type: str  # table, view, file, api, topic
    location: str
    schema: dict
    row_count: int
    owner: Optional[str] = None
    description: Optional[str] = None
    tags: list = field(default_factory=list)
    pii_columns: list = field(default_factory=list)
    sensitivity_level: str = "internal"
    last_updated: Optional[str] = None

class AIDataCatalog:
    def __init__(self):
        self.llm = Anthropic()
        self.assets = {}

    def scan_database(self, connection_string: str,
                       database_name: str) -> list[DataAsset]:
        """Сканирование базы данных и создание assets"""
        engine = sqlalchemy.create_engine(connection_string)
        inspector = sqlalchemy.inspect(engine)
        assets = []

        for table_name in inspector.get_table_names():
            columns = inspector.get_columns(table_name)
            schema = {col['name']: str(col['type']) for col in columns}

            # Получение семпла данных
            try:
                sample_df = pd.read_sql(f"SELECT * FROM {table_name} LIMIT 5", engine)
                sample_data = sample_df.to_dict('records')
                row_count = pd.read_sql(
                    f"SELECT COUNT(*) as cnt FROM {table_name}", engine
                )['cnt'].iloc[0]
            except Exception:
                sample_data = []
                row_count = 0

            # AI-классификация
            classification = self._classify_asset(
                table_name, schema, sample_data, database_name
            )

            asset = DataAsset(
                asset_id=f"{database_name}.{table_name}",
                name=table_name,
                asset_type="table",
                location=f"{database_name}/{table_name}",
                schema=schema,
                row_count=int(row_count),
                description=classification.get('description'),
                tags=classification.get('tags', []),
                pii_columns=classification.get('pii_columns', []),
                sensitivity_level=classification.get('sensitivity_level', 'internal')
            )

            assets.append(asset)
            self.assets[asset.asset_id] = asset

        return assets

    def _classify_asset(self, table_name: str, schema: dict,
                         sample_data: list, context: str = "") -> dict:
        """LLM-классификация датасета"""
        schema_str = json.dumps(schema)
        sample_str = json.dumps(sample_data[:3], ensure_ascii=False)[:500]

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"""Classify this database table for a data catalog.

Table: {table_name}
Database context: {context}
Schema: {schema_str}
Sample data: {sample_str}

Return JSON:
{{
  "description": "Brief business description of what this table contains",
  "domain": "business domain (e.g., users, orders, payments, analytics, logs)",
  "tags": ["tag1", "tag2"],
  "pii_columns": ["columns containing personal data"],
  "sensitivity_level": "public|internal|confidential|restricted",
  "data_category": "master|transactional|analytical|operational|reference"
}}"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {'description': 'Auto-discovered table', 'tags': [], 'pii_columns': []}

Поиск по каталогу

    def search(self, query: str, filters: dict = None) -> list[DataAsset]:
        """Семантический поиск по каталогу"""
        # Подготовка описаний всех активов
        asset_descriptions = []
        for asset_id, asset in self.assets.items():
            desc = f"{asset.name}: {asset.description or 'No description'}"
            desc += f" Tags: {', '.join(asset.tags)}"
            desc += f" Columns: {', '.join(list(asset.schema.keys())[:10])}"
            asset_descriptions.append({'id': asset_id, 'description': desc})

        # LLM ищет релевантные активы
        descriptions_text = "\n".join([
            f"{i+1}. {a['id']}: {a['description']}"
            for i, a in enumerate(asset_descriptions[:50])
        ])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Find relevant data assets for this query.

Query: {query}

Available assets:
{descriptions_text}

Return comma-separated IDs of relevant assets (top 5). No explanation."""
            }]
        )

        relevant_ids = [id.strip() for id in response.content[0].text.split(',')]

        results = [self.assets[id] for id in relevant_ids if id in self.assets]

        # Применение фильтров
        if filters:
            if 'sensitivity_level' in filters:
                results = [r for r in results
                           if r.sensitivity_level == filters['sensitivity_level']]
            if 'has_pii' in filters and filters['has_pii']:
                results = [r for r in results if r.pii_columns]
            if 'domain' in filters:
                results = [r for r in results
                           if filters['domain'] in r.tags]

        return results

    def find_related_assets(self, asset_id: str) -> list[dict]:
        """Поиск связанных датасетов по семантическому сходству"""
        if asset_id not in self.assets:
            return []

        source_asset = self.assets[asset_id]

        # Описания всех других активов
        other_assets = {id: asset for id, asset in self.assets.items() if id != asset_id}
        others_desc = "\n".join([
            f"- {id}: {asset.description}, columns: {list(asset.schema.keys())[:5]}"
            for id, asset in list(other_assets.items())[:30]
        ])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Find assets related to:
{source_asset.name}: {source_asset.description}
Columns: {list(source_asset.schema.keys())}

Other assets:
{others_desc}

Return JSON array: [{{"id": "...", "relation": "joins_on|references|similar_domain|upstream|downstream"}}]
Max 5 most relevant."""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return []

    def generate_data_dictionary(self, asset_id: str) -> dict:
        """Автогенерация data dictionary для датасета"""
        if asset_id not in self.assets:
            return {}

        asset = self.assets[asset_id]

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=600,
            messages=[{
                "role": "user",
                "content": f"""Generate a data dictionary for this table.

Table: {asset.name}
Description: {asset.description}
Schema: {json.dumps(asset.schema)}

Return JSON: {{"column_name": {{"description": "...", "example": "...", "notes": "..."}}}}"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {}

Как детектируем PII с точностью 99%?

Модель получает DDL и семпл — до 5 строк на таблицу. LLM распознает паттерны: номера телефонов, email, паспортные данные, медицинские коды. Каждая колонка проверяется на соответствие более 50 шаблонам PII. Результат — список PII-колонок с уровнем уверенности. Дополнительно запускается эвристический валидатор на регулярных выражениях: если LLM не уверена, а регэксп находит совпадение — колонка помечается как подозрительная. Такой гибридный подход дает recall >99% и precision 95%.

    def audit_pii_exposure(self) -> dict:
        """Аудит PII данных по всему каталогу"""
        pii_report = {
            'total_assets': len(self.assets),
            'assets_with_pii': [],
            'pii_columns_by_type': {}
        }

        for asset_id, asset in self.assets.items():
            if asset.pii_columns:
                pii_report['assets_with_pii'].append({
                    'asset': asset_id,
                    'pii_columns': asset.pii_columns,
                    'sensitivity': asset.sensitivity_level,
                    'owner': asset.owner
                })

        return pii_report

Чек-лист внедрения:

• Аудит 5-10 источников
• Выбор LLM и векторной БД
• Настройка коннекторов
• Разработка кастомных классификаторов
• Пилот на 50 активах (2 недели)
• Валидация точности и дообучение
• Развертывание в production
• Обучение операторов

OpenMetadata и DataHub — наиболее зрелые open-source решения для корпоративного каталога. AI-слой поверх них добавляет автоматическую классификацию при обнаружении новых таблиц: тегирование занимает 30-60 секунд вместо ручного заполнения за 15-30 минут на актив. Для организации с 500+ таблицами это экономит 100-200 часов при первоначальном заполнении каталога. Как отметил один из наших клиентов, «автоматическая классификация сократила время поиска данных с 40 минут до 5 секунд».

Процесс внедрения

1. Аналитика: аудит текущих источников данных, выбор коннекторов, определение доменов и чувствительности.
2. Проектирование: настройка схемы каталога, интеграция с IAM, выбор LLM и векторной базы.
3. Реализация: разработка коннекторов, кастомных классификаторов, семантического поиска.
4. Тестирование: пилот на 50 активах, валидация точности, дообучение моделей.
5. Деплой: развёртывание в production, обучение команды, Documentation.

Сроки внедрения

Что входит в стоимость?

Базовый пакет включает интеграцию с 1-2 источниками, AI-классификацию, семантический поиск и отчёт по PII. Расширенный пакет добавляет lineage, кастомные теги, ролевую модель и IAM. Сопровождение — мониторинг и дообучение каждые 6 месяцев. Точную стоимость рассчитываем индивидуально по числу активов и источников.

Получите консультацию: оценим ваш стек за 1 день, предложим пилот. Закажите внедрение — мы гарантируем точность классификации 95%+, иначе доработаем за свой счёт. Свяжитесь с нами для обсуждения деталей вашего проекта.

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

ETL для аналитики и ETL для ML — разные задачи. В аналитике важна агрегация, в ML — индивидуальные записи с историей. В аналитике train/val/test split не нужен, в ML — критичен. В аналитике skew данных мешает интерпретации, в ML — напрямую влияет на качество модели.

Инструменты. Apache Spark (Wikipedia) для больших объёмов (10GB+): PySpark с DataFrames, оптимизации через partitioning и caching. dbt для трансформаций поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версионируется, тестируется. Pandas + Polars для объёмов до нескольких GB — Polars в 5-10x быстрее Pandas на типичных трансформациях.

Temporal splits. Для ML важно, что split по времени, а не случайный. Если данные временные (транзакции, события пользователей), случайный split даёт data leakage: модель видит «будущие» данные при обучении. Правило: train на периоде T1-T2, validation на T2-T3 (с gap для предотвращения leakage), test на T3-T4. Неправильный split может стоить 10–15% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. Delta Lake или Apache Iceberg — форматы с ACID-транзакциями, Change Data Capture, time travel.

Как избежать training-serving skew с помощью Feature Store

Feature Store решает проблему рассинхронизации между обучением и инференсом. Самая коварная ошибка в ML-инфраструктуре — training-serving skew: признак считается по-разному в обучении и в продакшене. Модель учится на «правильных» данных, а инференс получает другие.

Feast (open source) — офлайн store на Parquet/Delta в S3 для обучения, онлайн store на Redis для low-latency инференса (<10ms). Feature definitions как Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition, используется везде. Нет расхождений.

Потоковые признаки. Когда признак должен обновляться в реальном времени (количество транзакций за последние 10 минут), нужна потоковая обработка. Apache Kafka + Apache Flink или Kafka Streams для вычисления признаков в реальном времени → запись в онлайн store. Сложнее, дороже, нужно только когда staleness признаков критична для качества.

Разметка данных: как не потратить бюджет впустую

Разметка — самая трудоёмкая и недооцениваемая часть ML-проекта. Плохо размеченные данные не исправит никакая архитектура.

Label Studio — open source, поддерживает разметку изображений (bounding box, polygon, segmentation), текста (NER, классификация), аудио, видео. Поднимается за 10 минут через Docker. Для небольших команд — первый выбор.

Оценка качества разметки. Inter-annotator agreement — насколько согласны разметчики между собой. Cohen's Kappa > 0.8 — хорошо, 0.6-0.8 — приемлемо, < 0.6 — задача неоднозначна или инструкция плохая. Пересечение разметок (10-20% примеров размечают два независимых аннотатора) — обязательная практика.

Active learning. Не размечать случайные примеры, а выбирать те, на которых модель наиболее неуверена (low confidence, high uncertainty). Позволяет добиться того же качества при 50-70% объёма разметки. Modals, Prodigy, Label Studio поддерживают active learning workflows. На одном из проектов для NLP мы сократили бюджет на разметку в 2,5 раза за счёт active learning.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для CV: рендеринг в Blender/Unity с реалистичными текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Риск: модель обучается на distribution синтетических данных, а не реальных — нужна осторожность и проверка на реальном holdout.

Качество данных: валидация и мониторинг

Great Expectations — de facto стандарт для data validation в ML-пайплайнах. Expectations — это декларативные утверждения о данных: «колонка age содержит значения от 0 до 120», «колонка user_id не содержит null», «распределение amount не отклоняется более чем на 20% от baseline». Запускается в пайплайне, при провале — блокирует прохождение.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation с type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель ожидает данные за последние N дней. ETL упал, данные не обновились — модель использует устаревшие признаки. Мониторинг свежести данных: timestamp последней записи в каждой таблице, алерт при задержке > порога.

Дедупликация. Дубликаты в обучающей выборке завышают метрики (одни и те же примеры в train и val) и искажают веса модели. MinHash LSH для приближённой дедупликации больших датасетов. Для точной — хэш по нормализованному контенту.

Инструменты валидации: сравнение

Хранилища и форматы

Для большинства ML-проектов на старте: Parquet в S3 + DVC для версионирования. Delta Lake или Iceberg — когда появляется потребность в инкрементальных обновлениях или time travel.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

• Аудит существующих данных и пайплайнов (1 неделя).
• Проектирование архитектуры: выбор инструментов, форматов, способов разметки.
• Реализация ETL/ELT пайплайна с валидацией и мониторингом.
• Документация кода и процессов (model card, data card).
• Обучение вашей команды работе с пайплайном.
• SLA на сопровождение и поддержку.

Как мы строим пайплайн: пошагово

1. Аудит существующих данных. Профилирование: ydata-profiling (бывший pandas-profiling) генерирует HTML-репорт со статистиками, дистрибуциями, корреляциями, missing values за минуты.
2. Проектирование пайплайна. Определяем источники данных, частоту обновления, требования к latency признаков, объёмы.
3. Реализация и тестирование. Unit-тесты на трансформации, integration-тесты на пайплайн, data validation через Great Expectations.
4. Деплой и мониторинг. Алерты на freshness, quality checks, аномалии в объёмах данных.

Почему стоит доверить это нам

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

Свяжитесь с нами для бесплатного аудита ваших данных — оценим текущий пайплайн и предложим roadmap. Закажите построение ML-пайплайна под ключ.