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-решений. Гарантируем целостность данных — если после миграции обнаружится расхождение, исправляем за свой счёт. Получите консультацию: мы оценим вашу задачу и предложим оптимальный подход.