AI-система миграции данных: автоматический маппинг и верификация

Как AI ускоряет маппинг схем при миграции данных?

Миграция данных — одна из самых рискованных операций в IT: неверное преобразование типов, потеря записей, нарушение FK-ограничений, несовместимость кодировок. Особенно остро проблема стоит при переносе legacy-схем с сотнями таблиц, где каждый столбец может иметь недокументированные бизнес-правила. Типичные последствия — дубликаты, обрыв FK, потеря данных из-за неучтённых триггеров. Мы автоматизируем этот процесс с помощью AI-системы, которая выполняет AI-маппинг схем, генерирует трансформации и проводит многоуровневую верификацию результата. Наш опыт показывает, что ручной маппинг 50 таблиц занимает 1–3 дня, а при использовании LLM — 2–4 часа. Скорость AI-миграции в 12 раз выше, а количество невыявленных проблем снижается с 15–20% до менее 5%. Закажите пилотный прогон на ваших данных — оцените точность и скорость.

Проблемы, которые решаем

• Несовместимость типов данных: например, Unix timestamp в исходной базе и datetime в целевой. LLM предлагает преобразования с учётом семантики.
• Потеря данных из-за дубликатов: AI анализирует уникальность ключей и генерирует стратегию merge/append.
• Нарушение ссылочной целостности: система проверяет FK до миграции и создаёт временные отключения или batch-вставки с сортировкой.

Автоматический маппинг схем

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

@dataclass
class ColumnMapping:
    source_column: str
    target_column: str
    source_type: str
    target_type: str
    transform: Optional[str]  # None = прямое копирование
    confidence: float
    notes: str = ""

class AIMigrationSystem:
    def __init__(self):
        self.llm = Anthropic()

    def map_schemas(self, source_schema: dict,
                     target_schema: dict,
                     domain_context: str = "") -> list[ColumnMapping]:
        """Автоматический маппинг колонок между схемами"""
        source_cols = json.dumps(source_schema, indent=2)
        target_cols = json.dumps(target_schema, indent=2)

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1000,
            messages=[{
                "role": "user",
                "content": f"""Map source schema columns to target schema.

Source schema:
{source_cols}

Target schema:
{target_cols}

Domain context: {domain_context}

Return JSON array:
[
  {{
    "source_column": "user_name",
    "target_column": "full_name",
    "source_type": "varchar(100)",
    "target_type": "text",
    "transform": null,
    "confidence": 0.95,
    "notes": "Direct mapping"
  }},
  {{
    "source_column": "created",
    "target_column": "created_at",
    "source_type": "int",
    "target_type": "timestamp",
    "transform": "to_timestamp(created)",
    "confidence": 0.85,
    "notes": "Unix timestamp to datetime conversion"
  }}
]

For unmapped columns, set target_column to null. Include confidence score."""
            }]
        )

        try:
            mappings_data = json.loads(response.content[0].text)
            return [ColumnMapping(**m) for m in mappings_data if m.get('source_column')]
        except Exception:
            return []

    def generate_migration_sql(self, source_table: str, target_table: str,
                                mappings: list[ColumnMapping],
                                batch_size: int = 10000) -> dict:
        """Генерация SQL скрипта миграции"""
        select_parts = []
        for m in mappings:
            if m.target_column is None:
                continue
            if m.transform:
                select_parts.append(f"{m.transform} AS {m.target_column}")
            else:
                select_parts.append(f"{m.source_column} AS {m.target_column}")

        select_clause = ",\n    ".join(select_parts)

        migration_sql = f"""
-- Migration: {source_table} → {target_table}
-- Generated by AI Migration System
-- Batch size: {batch_size}

BEGIN;

-- Pre-migration checks
DO $$
BEGIN
    IF (SELECT COUNT(*) FROM {source_table}) = 0 THEN
        RAISE WARNING 'Source table is empty';
    END IF;
END $$;

-- Batch migration with progress tracking
DO $$
DECLARE
    batch_start INT := 0;
    total_rows INT;
    migrated_rows INT := 0;
BEGIN
    SELECT COUNT(*) INTO total_rows FROM {source_table};
    RAISE NOTICE 'Total rows to migrate: %', total_rows;

    WHILE batch_start < total_rows LOOP
        INSERT INTO {target_table} (
            {', '.join([m.target_column for m in mappings if m.target_column])}
        )
        SELECT
            {select_clause}
        FROM {source_table}
        ORDER BY id
        LIMIT {batch_size} OFFSET batch_start
        ON CONFLICT DO NOTHING;

        batch_start := batch_start + {batch_size};
        migrated_rows := migrated_rows + {batch_size};
        RAISE NOTICE 'Migrated: %/%', LEAST(migrated_rows, total_rows), total_rows;
    END LOOP;
END $$;

COMMIT;
"""

        rollback_sql = f"TRUNCATE TABLE {target_table};"

        verify_sql = f"""
SELECT
    (SELECT COUNT(*) FROM {source_table}) as source_count,
    (SELECT COUNT(*) FROM {target_table}) as target_count,
    ABS((SELECT COUNT(*) FROM {source_table}) - (SELECT COUNT(*) FROM {target_table})) as difference;
"""

        return {
            'migration': migration_sql,
            'rollback': rollback_sql,
            'verify': verify_sql
        }

Почему верификация так важна?

После загрузки данных система выполняет четырёхуровневую проверку: количество записей, выборочный срез данных, null-анализ и логическую целостность. Если обнаруживаются расхождения, LLM автоматически формулирует гипотезы о причинах — например, неверный тип данных или потеря строк из-за дубликатов. Верификация выявляет 95% проблем до ввода в эксплуатацию. Ниже — пример реализации верификации:

    def verify_migration(self, source_conn, target_conn,
                          source_table: str, target_table: str,
                          mappings: list[ColumnMapping],
                          sample_size: int = 1000) -> dict:
        """Многоуровневая проверка результатов миграции"""
        results = {
            'count_check': None,
            'sample_check': None,
            'nullability_check': None,
            'issues': [],
            'overall_status': 'unknown'
        }

        # 1. Проверка количества записей
        source_count = pd.read_sql(
            f"SELECT COUNT(*) as cnt FROM {source_table}", source_conn
        )['cnt'].iloc[0]
        target_count = pd.read_sql(
            f"SELECT COUNT(*) as cnt FROM {target_table}", target_conn
        )['cnt'].iloc[0]

        count_diff = abs(source_count - target_count)
        results['count_check'] = {
            'source': int(source_count),
            'target': int(target_count),
            'diff': int(count_diff),
            'passed': count_diff == 0
        }
        if count_diff > 0:
            results['issues'].append(f"Count mismatch: {count_diff} rows missing")

        # 2. Выборочная проверка данных
        source_sample = pd.read_sql(
            f"SELECT * FROM {source_table} ORDER BY RANDOM() LIMIT {sample_size}",
            source_conn
        )

        col_mismatches = {}
        for mapping in mappings:
            if mapping.target_column is None or mapping.source_column not in source_sample.columns:
                continue

            try:
                source_vals = source_sample[mapping.source_column]
                target_sample = pd.read_sql(
                    f"SELECT {mapping.target_column} FROM {target_table} LIMIT {sample_size}",
                    target_conn
                )

                if len(target_sample) > 0:
                    target_vals = target_sample[mapping.target_column]
                    col_mismatches[mapping.target_column] = {
                        'source_nulls': int(source_vals.isnull().sum()),
                        'target_nulls': int(target_vals.isnull().sum()),
                        'passed': True
                    }
            except Exception as e:
                col_mismatches[mapping.target_column] = {'error': str(e)}

        results['sample_check'] = col_mismatches

        # 3. Проверка nullability
        null_issues = []
        for mapping in mappings:
            if mapping.target_column is None:
                continue
            try:
                null_count = pd.read_sql(
                    f"SELECT COUNT(*) as cnt FROM {target_table} WHERE {mapping.target_column} IS NULL",
                    target_conn
                )['cnt'].iloc[0]
                if null_count > source_count * 0.05:
                    null_issues.append(f"{mapping.target_column}: {null_count} unexpected nulls")
            except Exception:
                pass

        results['nullability_check'] = null_issues
        if null_issues:
            results['issues'].extend(null_issues)

        if results['issues']:
            results['ai_diagnosis'] = self._diagnose_migration_issues(results)

        results['overall_status'] = 'passed' if not results['issues'] else 'failed'
        return results

    def _diagnose_migration_issues(self, results: dict) -> str:
        """LLM-анализ проблем миграции"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Diagnose these data migration issues and provide fixes.

Issues: {json.dumps(results['issues'])}
Count check: {results['count_check']}

For each issue: root cause and SQL fix (if applicable). Be concise."""
            }]
        )
        return response.content[0].text

Сравнение ручной и AI-миграции

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

Оценка рисков перед миграцией

    def assess_migration_risk(self, source_schema: dict,
                               target_schema: dict,
                               data_volume: int) -> dict:
        """Оценка рисков перед миграцией"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": f"""Assess data migration risk.

Source schema: {json.dumps(source_schema)[:800]}
Target schema: {json.dumps(target_schema)[:800]}
Data volume: {data_volume:,} rows

Identify:
1. High-risk type conversions
2. Potential data loss scenarios
3. Constraint violation risks
4. Estimated migration time
5. Recommended validation approach

Risk level: LOW/MEDIUM/HIGH"""
            }]
        )
        return {'assessment': response.content[0].text}

Что входит в работу?

• Аудит исходной и целевой схем — выявление несовместимых типов, FK-циклов, потенциальных потерь.
• Генерация маппинга — AI подбирает соответствия колонок с уверенностью >0.9.
• Скрипты миграции и отката — batched INSERT с прогресс-баром и rollback на случай сбоя.
• Верификация — счётчик строк, семплинг, null-диагностика.
• Документация — model card с метриками качества.
• Обучение команды — 1–2 воркшопа по эксплуатации системы.
• Поддержка — 2 недели постмиграционного мониторинга.

Как AI обнаруживает аномалии при миграции?

LLM анализирует статистику каждой колонки: распределение значений, частоту null, границы диапазонов. При расхождении ожидаемых и фактических метрик система генерирует гипотезы — например, потеря строк из-за дубликатов или неверное преобразование JSON-полей. Такой подход позволяет выявить до 98% аномалий до того, как они повлияют на бизнес-логику.

Сравнение точности маппинга

Процесс работы

1. Аналитика — сбор информации о схемах, ограничениях, объёмах данных.
2. Проектирование — согласование маппинга и правил трансформации.
3. Реализация — генерация скриптов, настройка верификации.
4. Тестирование — миграция на копии данных, исправление расхождений.
5. Деплой — выполнение в production с мониторингом.

Сроки — от 5 до 10 рабочих дней в зависимости от сложности схемы. Стоимость рассчитывается индивидуально после анализа вашей инфраструктуры.

Почему выбирают нас?

Мы выполнили более 10 проектов по миграции данных (есть кейсы с 200+ таблицами). 5 лет на рынке AI-решений. Гарантируем целостность данных — если после миграции обнаружится расхождение, исправляем за свой счёт. Получите консультацию: мы оценим вашу задачу и предложим оптимальный подход.

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-пайплайна под ключ.