AI-каталог даних з автокласифікацією та PII-детекцією

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

Наша компанія має 5+ років досвіду впровадження data governance рішень та понад 30 успішних проектів для фінтех, ритейлу та телекому.

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

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

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

AI-каталог швидший за ручний у 25 разів, а точність вища на 20-30%.

Як ми будуємо AI-каталог?

Сканування бази через SQLAlchemy, вилучення семплу, LLM-класифікація — стек на Python з PyTorch для embeddings (all-MiniLM-L6-v2), ChromaDB для векторного пошуку. Підтримуємо реляційні бази (PostgreSQL, MySQL, ClickHouse), файлові сховища (S3, HDFS), Kafka-топіки та REST API. Приклад: скануємо 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

Після калібрування під домен точність досягає 95–98%, а для PII-колонок recall перевищує 99%.

Процес впровадження

1. Аналітика: аудит поточних джерел даних, вибір конекторів, визначення доменів та чутливості.
2. Проектування: налаштування схеми каталогу, інтеграція з IAM (LDAP, AD), вибір LLM та векторної бази.
3. Реалізація: розробка конекторів, кастомних класифікаторів, семантичного пошуку.
4. Тестування: пілот на 50 активах, валідація точності, донавчання моделей.
5. Деплой: розгортання в production, навчання команди, документація.

Інтегрується з OpenMetadata або DataHub, Slack-сповіщення, експорт у dbt docs.

Строки впровадження

Що входить у вартість?

Базовий пакет (від $10,000) включає інтеграцію з 1-2 джерелами, AI-класифікацію, семантичний пошук та звіт з PII. Розширений пакет (від $25,000) додає лайн-лінію, кастомні теги, рольову модель та IAM. Супровід — моніторинг та донавчання кожні 6 місяців. Точну вартість розраховуємо індивідуально за кількістю активів та джерел.

Отримайте консультацію: оцінимо ваш стек за 1 день, запропонуємо пілот. Проводимо двотижневий пілот на 50 таблицях — ви оцінюєте точність до покупки. Замовте впровадження — ми гарантуємо точність класифікації 95%+, інакше доопрацюємо за свій рахунок. Зв'яжіться з нами для обговорення деталей вашого проекту.

Чому дата-інжиніринг визначає успіх ML-моделі

Минулого року до нас звернулася компанія, яка витратила $50 000 на навчання NLP-моделі, але отримала лише 60% точності на продакшені. Причина — data leakage через випадковий split часових даних. Перед тим як навчати модель, потрібно зрозуміти структуру даних: чи є дублі, як часто змінюється схема, наскільки репрезентативна вибірка. Дата-інжиніринг для ML — це не просто ETL, а побудова відтворюваної інфраструктури, яка робить навчання надійним, а перенавчання — передбачуваним. За досвідом нашої команди (понад 8 років у дата-інжинірингу, 30+ проектів у ML) кожна друга проблема в продакшені пов’язана не з архітектурою моделі, а з якістю даних. Замовте аудит ваших даних — оцінимо поточний пайплайн безкоштовно.

Як ETL-пайплайни для ML відрізняються від BI

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

Інструменти. Apache Spark для великих обсягів (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% якості моделі на валідації. Temporal split best practices (scikit-learn docs)

Інкрементальні пайплайни. Модель перенавчається щотижня на нових даних. Потрібен пайплайн, який інкрементально додає нові записи до навчальної вибірки, не перевантажуючи все з нуля. 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 — економія склала $15 000 на 100 000 розмічених прикладів.

Синтетичні дані. Коли реальних даних мало або отримати їх дорого. Для 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.

Типові помилки при побудові пайплайнів

• Пропуск перевірки свіжості даних. Якщо ETL падає вночі, а модель запускається вранці — вона отримує дані 24-годинної давності. Рішення: алерт при затримці > 30 хвилин.
• Відсутність версіонування даних. Не можна відтворити експеримент, бо дані змінилися. DVC або Delta Lake time travel виправляють це.
• Забувають про schema evolution. Нове поле з’являється, а пайплайн падає. Автоматичне виявлення змін схеми через Great Expectations.

Active learning дозволяє скоротити бюджет на розмітку до 50–70%. На одному проекті це склало економію $15 000 на 100 000 розмічених прикладів. Закажіть консультацію — розрахуємо потенційну економію для вашого кейсу.

Що входить у проект з дата-інжинірингу для ML

Ми надаємо повний цикл:

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

Терміни: від 2 до 6 тижнів залежно від обсягу даних і складності інтеграцій.

Як ми будуємо пайплайн: покроково

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

Чому варто довірити це нам

Ми займаємося дата-інжинірингом та ML з понад 8-річним досвідом. За цей час реалізували понад 40 проектів — від побудови пайплайнів для NLP-моделей до розмітки датасетів для комп’ютерного зору. Гарантуємо відтворюваність пайплайнів та повну прозорість процесів. У кожному проекті використовуємо інструменти з відкритим кодом, щоб ви не були прив’язані до вендора.

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