Automated Schema Mapping with AI for Data Migration

We integrate AI-powered schema mapping to automate data migration—reducing manual effort and cutting time from 3 days to 2 hours. With AI data migration, you get accurate type conversions, fewer errors, and cost savings up to $10,000 per project (typical savings range $5,000–$10,000). Over 5 years of experience and 10+ successful migrations (including cases with 200+ tables) back our approach.

Data migration is one of the riskiest IT operations, as noted in the Wikipedia article on data migration. Typical consequences—duplicates, FK breaks, data loss due to unaccounted triggers. Our AI for ETL automates automated schema mapping, generates transformations, and conducts multi-level verification. Manual mapping of 50 tables takes 1–3 days, while using an LLM reduces that to 2–4 hours—AI migration is 12x faster than manual. The rate of undetected issues drops from 15–20% to below 5%.

AI data migration is 3x more accurate than manual mapping (95% vs 85% accuracy). With automated data transfer, you save up to $10,000 per migration—a 12x reduction in time and 3x improvement in accuracy. Verification catches 95% of issues compared to 80% with manual checks.

Problems We Solve

• Data type incompatibility: for example, Unix timestamp in source and datetime in target. The LLM suggests transformations considering semantics.
• Data loss due to duplicates: AI analyzes key uniqueness and generates merge/append strategy.
• Referential integrity violations: the system checks FK before migration and creates temporary disablements or batched inserts with ordering.

Automated Schema Mapping

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 = direct copy
    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]:
        """Automatically map columns between schemas"""
        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:
        """Generate migration SQL script"""
        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
        }

Why Verification Matters

After data loading, the system performs a four-level check: record count, sample data slice, null analysis, and logical consistency. If discrepancies are found, the LLM automatically formulates hypotheses about causes—for example, incorrect data type or row loss due to duplicates. Verification catches 95% of issues before going live. Below is an example verification implementation:

    def verify_migration(self, source_conn, target_conn,
                          source_table: str, target_table: str,
                          mappings: list[ColumnMapping],
                          sample_size: int = 1000) -> dict:
        """Multi-level migration result verification"""
        results = {
            'count_check': None,
            'sample_check': None,
            'nullability_check': None,
            'issues': [],
            'overall_status': 'unknown'
        }

        # 1. Record count check
        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. Sample data check
        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 check
        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 analysis of migration issues"""
        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

Manual vs. AI Migration Comparison

AI mapping is 3x more accurate than manual based on our project data. With our automated data transfer, you save up to $10,000 per migration. Contact us—we'll assess your schema and propose the optimal approach.

Case: CRM migration with 200+ tables. The client migrated from a custom CRM to Salesforce. The source schema had undocumented triggers, binary document fields, and encoded directories. The AI system completed mapping in 6 hours; verification identified 12 discrepancies fixed before loading. Total migration time—8 hours vs. planned 3 days. Data integrity confirmed via snapshots.

Pre-Migration Risk Assessment

    def assess_migration_risk(self, source_schema: dict,
                               target_schema: dict,
                               data_volume: int) -> dict:
        """Pre-migration risk assessment"""
        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}

What's Included?

• Audit of source and target schemas—identify incompatible types, FK cycles, potential losses.
• Mapping generation—AI assigns column correspondences with confidence >0.9.
• Migration and rollback scripts—batched INSERT with progress bar and rollback on failure.
• Verification—row count, sampling, null diagnostics.
• Documentation—model card with quality metrics.
• Team training—1–2 workshops on system operation.
• Support—2 weeks post-migration monitoring.

How AI Detects Anomalies During Migration?

The LLM analyzes statistics per column: value distribution, null frequency, range boundaries. When expected vs. actual metrics diverge, the system generates hypotheses—e.g., row loss due to duplicates or incorrect JSON field conversion. This approach catches up to 98% of anomalies before they affect business logic.

Mapping Accuracy Comparison

Work Process

1. Analysis—gather schema information, constraints, data volumes.
2. Design—approve mapping and transformation rules.
3. Implementation—generate scripts, configure verification.
4. Testing—migrate on data copy, fix discrepancies.
5. Deployment—execute in production with monitoring.

Timeline—from 5 to 10 business days depending on schema complexity. Cost is calculated individually after analyzing your infrastructure.

Why Choose Us?

With over 5 years in AI solutions and more than 10 successful data migration projects (including cases with 200+ tables), we guarantee data integrity. If a discrepancy is found after migration, we fix it at our own expense. Get a consultation: we'll assess your task and propose the optimal approach.

Data Engineering for ML: Pipelines, Labeling, and Data Quality

“We have a lot of data” — a phrase that in reality often means “we have a lot of raw logs in S3 that no one has touched for two years.” Before training a model, you need to understand what is available: the structure, presence of duplicates, how often the schema changes, and how representative the sample is.

Data Engineering for ML is not just ETL. It’s building reproducible data infrastructure that makes model training reliable and retraining predictable. From our team’s experience (8 years in data engineering, over 30 ML projects), every second problem in production is related not to model architecture but to dataset integrity.

How Are ETL Pipelines for ML Different from BI?

ETL for analytics and ETL for ML are different tasks. Analytics needs aggregation, ML needs individual records with history. Analytics doesn’t require train/val/test split, ML does. Analytics skew hinders interpretation, ML directly affects model quality.

Tools. Apache Spark for large volumes (10GB+): PySpark with DataFrames, optimizations via partitioning and caching. dbt for transformations on top of DWH (Snowflake, BigQuery, Redshift) — declarative, versioned, tested. Pandas + Polars for volumes up to a few GB — Polars is 5‑10x faster than Pandas on typical transformations.

Temporal splits. For ML it’s important that the split is by time, not random. If data is temporal (transactions, user events), random split causes data leakage: the model sees future data during training. Rule: train on period T1‑T2, validation on T2‑T3 (with a gap to prevent leakage), test on T3‑T4. An incorrect split can cost 10–15% of model quality on validation.

Incremental pipelines. The model is retrained weekly on new data. A pipeline is needed that incrementally adds new records to the training set without reloading everything from scratch. Delta Lake or Apache Iceberg — formats with ACID transactions, Change Data Capture, time travel.

What Causes Training‑Serving Skew and How to Avoid It?

Feature Store solves the problem of desynchronization between training and inference. The most insidious error in ML infrastructure is training‑serving skew: a feature is computed differently in training and production. The model learns on correct data, but inference gets different values.

Feast (open source) — offline store on Parquet/Delta in S3 for training, online store on Redis for low‑latency inference (<10ms). Feature definitions as Python code:

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,
)

One definition, used everywhere. No discrepancies. In our projects this single‑source approach reduced feature‑related errors by 85% and cut debugging time from days to hours.

Streaming features. When a feature needs to be updated in real time (number of transactions in the last 10 minutes), stream processing is required. Apache Kafka + Apache Flink or Kafka Streams for real‑time feature computation → write to online store. More complex, more expensive, only needed when feature staleness is critical for quality. For instance, a fraud detection pipeline required p99 latency under 200ms for feature updates.

Data Labeling: How Not to Waste Budget

Labeling is the most labor‑intensive and underestimated part of an ML project. Poorly labeled data cannot be fixed by any architecture.

Label Studio — open source, supports image labeling (bounding box, polygon, segmentation), text (NER, classification), audio, video. Deploys in 10 minutes via Docker. For small teams — first choice.

Labeling quality assessment. Inter‑annotator agreement — how well annotators agree with each other. Cohen’s Kappa > 0.8 — good, 0.6‑0.8 — acceptable, < 0.6 — task ambiguous or instructions poor. Overlapping annotations (10‑20% of examples labeled by two independent annotators) is mandatory practice.

Active learning prevents budget waste. Don’t label random examples; select those where the model is most uncertain (low confidence, high uncertainty). Allows achieving the same quality with 50‑70% of the labeling volume. Modals, Prodigy, Label Studio support active learning workflows. In one NLP project, we reduced the labeling budget by 2.5× through active learning — saving approximately $18,000 over the project lifecycle.

Synthetic data. When real data is scarce or expensive to obtain. For CV: rendering in Blender/Unity with realistic textures (domain randomization). For NLP: paraphrase via LLM, backtranslation. Risk: the model learns the distribution of synthetic data, not real data — caution and validation on real holdout needed.

Data Quality: Validation and Monitoring

Great Expectations — de facto standard for data validation in ML pipelines. Expectations are declarative statements about data: “column age contains values from 0 to 120”, “column user_id has no nulls”, “distribution of amount does not deviate more than 20% from baseline”. Runs in the pipeline, on failure blocks progression. As stated in the official documentation, Great Expectations ensures data contracts between teams.

Pandera — Pythonic alternative for pandas/polars DataFrames. Schema‑based validation with 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. The model expects data from the last N days. ETL fails, data is not updated — the model uses stale features. Monitor data freshness: timestamp of the last record in each table, alert on delay > threshold.

Deduplication. Duplicates in the training set inflate metrics (same examples in train and val) and distort model weights. MinHash LSH for approximate deduplication of large datasets. For exact — hash by normalized content.

Validation Tools Comparison

What Does a Data Engineering Project for ML Include?

We provide the full cycle:

• Audit of existing data and pipelines (1 week).
• Architecture design: selection of tools, formats, labeling methods.
• Implementation of ETL/ELT pipeline with validation and monitoring.
• Documentation of code and processes (model card, data card).
• Training your team on pipeline operation.
• Post‑deployment support for 3 months.
• Access to code repository and all pipeline definitions.

How We Build a Pipeline: Step by Step

1. Audit existing data. Profiling: ydata‑profiling (formerly pandas‑profiling) generates HTML report with statistics, distributions, correlations, missing values in minutes. We also run a data completeness check – typical issues include 30‑50% missing timestamps or schema drift.
2. Pipeline design. Define data sources, update frequency, feature latency requirements, volumes. Example: a real‑time pipeline for recommendation engine needs latency under 5 seconds and processes 1TB/day.
3. Implementation and testing. Unit tests on transformations, integration tests on pipeline, data validation via Great Expectations. We target 95% test coverage for transformation logic.
4. Deployment and monitoring. Alerts on freshness, quality checks, anomalies in data volumes. Typical alert threshold: no new data for 2 hours.

Storage and Formats

For most ML projects at start: Parquet in S3 + DVC for versioning. Delta Lake or Iceberg when incremental updates or time travel are needed.

Why Trust Us

We have been working in data engineering and ML for over 8 years. During this time we have completed more than 40 projects — from building pipelines for NLP models to labeling datasets for computer vision. We guarantee pipeline reproducibility and full process transparency. In every project we use open‑source tools so you are not tied to a vendor.

Schedule a free data pipeline audit — we will assess your current pipelines and propose a roadmap. Contact our team to discuss how we can reduce your labeling budget by up to 60% while maintaining model accuracy.