Implementing AI Deduplication and Data Cleaning

A typical scenario: your CRM has accumulated duplicate client records—different managers entered the same contacts with typos and in various formats. As a result, emails go out twice, and analytics show a distorted funnel. We integrate AI deduplication that finds non-obvious duplicates: 'Ivan Ivanov' and 'I. Ivanov', 'LLC Romashka' and 'Romashka Ltd'. Fuzzy matching combined with ML classification identifies such pairs, and an LLM adds semantic understanding—which record to consider canonical and how to merge attributes. The outcome is a single clean dataset without information loss. According to a study by Gartner, poor data quality costs organizations an average of $12.9 million annually. AI deduplication can significantly reduce these losses. With over 5 years of experience and 20+ data cleaning projects, our team of certified AI specialists ensures high-quality results. Implementing AI deduplication can save your company up to $50,000 per year in manual data cleaning costs.

How AI Deduplication Works

Classic SQL DISTINCT is useless when duplicates are not identical. We use a multi-level approach:

• Blocking: quick grouping of records by the first 3 characters of a normalized field.
• Fuzzy matching in Python: multiple similarity metrics (token sort, partial ratio) with weights and a bonus for matching email/phone.
• Clustering duplicates: Union-Find algorithm merges found pairs into duplicate groups.
• LLM conflict resolution: choosing the canonical record by the 'most complete record' rule or via LLM for complex cases; few-shot prompts improve accuracy.

For semantic deduplication, we use embeddings (1536-dim) and compare record vectors via cosine similarity. This pipeline processes 1M records in 15–30 minutes. At a threshold of 0.85, precision reaches ~92%, recall ~88%. LLM conflict resolution is applied only to 5–10% of pairs (the rest are merged automatically), keeping AI processing cost manageable.

Here is an example implementation:

import pandas as pd
import numpy as np
from anthropic import Anthropic
from rapidfuzz import fuzz, process
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
import re

class AIDeduplicator:
    def __init__(self, threshold: float = 0.85):
        self.llm = Anthropic()
        self.threshold = threshold
        self.classifier = None

    def deduplicate_persons(self, df: pd.DataFrame,
                             name_col: str,
                             email_col: str = None,
                             phone_col: str = None) -> pd.DataFrame:
        """De-duplicate persons with fuzzy matching"""
        # Normalize names
        df['_name_norm'] = df[name_col].apply(self._normalize_name)
        if email_col:
            df['_email_norm'] = df[email_col].apply(self._normalize_email)
        if phone_col:
            df['_phone_norm'] = df[phone_col].apply(self._normalize_phone)

        # Find duplicate pairs via blocking + fuzzy match
        duplicate_pairs = self._find_duplicate_pairs(df, name_col, email_col)

        # Build clusters of duplicates
        clusters = self._build_clusters(duplicate_pairs, len(df))

        # Pick canonical record in each cluster
        result = self._merge_clusters(df, clusters, name_col)

        return result

    def _normalize_name(self, name: str) -> str:
        if pd.isna(name):
            return ""
        name = str(name).lower().strip()
        name = re.sub(r'\s+', ' ', name)
        # Normalize Cyrillic/Latin
        name = re.sub(r'[^\w\s-]', '', name)
        return name

    def _normalize_email(self, email: str) -> str:
        if pd.isna(email):
            return ""
        return str(email).lower().strip()

    def _normalize_phone(self, phone: str) -> str:
        if pd.isna(phone):
            return ""
        # Keep only digits
        digits = re.sub(r'\D', '', str(phone))
        # Normalize Russian numbers
        if len(digits) == 11 and digits[0] == '8':
            digits = '7' + digits[1:]
        return digits[-10:] if len(digits) >= 10 else digits

    def _find_duplicate_pairs(self, df: pd.DataFrame,
                               name_col: str,
                               email_col: str = None) -> list[tuple]:
        """Find duplicate pairs via blocking"""
        pairs = []
        names = df['_name_norm'].tolist()

        # Quick blocking: first 3 letters of name
        blocks = {}
        for idx, name in enumerate(names):
            if len(name) >= 3:
                key = name[:3]
                if key not in blocks:
                    blocks[key] = []
                blocks[key].append(idx)

        # Fuzzy matching within blocks
        for block_indices in blocks.values():
            if len(block_indices) < 2:
                continue

            for i in range(len(block_indices)):
                for j in range(i + 1, len(block_indices)):
                    idx_a, idx_b = block_indices[i], block_indices[j]
                    name_a = names[idx_a]
                    name_b = names[idx_b]

                    # Multiple similarity metrics
                    ratio = fuzz.token_sort_ratio(name_a, name_b) / 100
                    partial = fuzz.partial_ratio(name_a, name_b) / 100
                    combined_score = (ratio * 0.7 + partial * 0.3)

                    # Bonus for matching email/phone
                    if email_col and '_email_norm' in df.columns:
                        email_a = df['_email_norm'].iloc[idx_a]
                        email_b = df['_email_norm'].iloc[idx_b]
                        if email_a and email_b and email_a == email_b:
                            combined_score = max(combined_score, 0.95)

                    if combined_score >= self.threshold:
                        pairs.append((idx_a, idx_b, combined_score))

        return pairs

    def _build_clusters(self, pairs: list[tuple],
                         total_records: int) -> list[list[int]]:
        """Union-Find to merge into clusters"""
        parent = list(range(total_records))

        def find(x):
            if parent[x] != x:
                parent[x] = find(parent[x])
            return parent[x]

        def union(x, y):
            px, py = find(x), find(y)
            if px != py:
                parent[px] = py

        for idx_a, idx_b, _ in pairs:
            union(idx_a, idx_b)

        # Group by root
        from collections import defaultdict
        clusters = defaultdict(list)
        for i in range(total_records):
            clusters[find(i)].append(i)

        # Only clusters with duplicates
        return [c for c in clusters.values() if len(c) > 1]

    def _merge_clusters(self, df: pd.DataFrame, clusters: list[list[int]],
                         name_col: str) -> pd.DataFrame:
        """Merge duplicates with canonical record selection"""
        rows_to_drop = set()
        updates = {}

        for cluster in clusters:
            cluster_df = df.iloc[cluster]

            # Heuristic: most complete = canonical
            completeness = cluster_df.notna().sum(axis=1)
            canonical_idx = cluster[completeness.argmax()]

            # LLM for complex cases (attribute conflicts)
            conflicts = self._detect_conflicts(cluster_df, name_col)
            if conflicts:
                resolution = self._resolve_conflicts_with_llm(cluster_df, conflicts)
                updates[canonical_idx] = resolution

            rows_to_drop.update(set(cluster) - {canonical_idx})

        # Apply updates and drop duplicates
        df_result = df.copy()
        for idx, update in updates.items():
            for col, val in update.items():
                if col in df_result.columns:
                    df_result.at[idx, col] = val

        df_result = df_result.drop(index=list(rows_to_drop)).reset_index(drop=True)

        return df_result

    def _detect_conflicts(self, cluster_df: pd.DataFrame,
                            name_col: str) -> dict:
        """Detect conflicting values in cluster"""
        conflicts = {}
        for col in cluster_df.columns:
            if col.startswith('_'):
                continue
            unique_vals = cluster_df[col].dropna().unique()
            if len(unique_vals) > 1:
                conflicts[col] = unique_vals.tolist()
        return conflicts

    def _resolve_conflicts_with_llm(self, cluster_df: pd.DataFrame,
                                     conflicts: dict) -> dict:
        """LLM picks correct value on conflict"""
        records = cluster_df.to_dict('records')
        records_str = json.dumps(records, ensure_ascii=False, default=str)

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""These are duplicate records that need to be merged.

Records:
{records_str[:800]}

Conflicting fields: {list(conflicts.keys())}

For each conflicting field, choose the most accurate/complete value.
Return JSON: {{"field_name": "chosen_value"}}"""
            }]
        )
        try:
            import json
            return json.loads(response.content[0].text)
        except Exception:
            return {}

Data Cleaning with LLM

LLMs can also standardize addresses and company names. Here is an example:

class AIDataCleaner:
    """Clean and standardize text fields"""

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

    def clean_addresses(self, addresses: list[str]) -> list[dict]:
        """Parse and standardize addresses"""
        batch_size = 10
        results = []

        for i in range(0, len(addresses), batch_size):
            batch = addresses[i:i + batch_size]
            addresses_str = "\n".join([f"{j+1}. {a}" for j, a in enumerate(batch)])

            response = self.llm.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=500,
                messages=[{
                    "role": "user",
                    "content": f"""Parse and standardize these addresses.

{addresses_str}

Return JSON array: [{{"original": "...", "country": "...", "city": "...", "street": "...", "building": "...", "apartment": "...", "postal_code": "..."}}]
Use null for missing fields."""
                }]
            )

            try:
                import json
                batch_results = json.loads(response.content[0].text)
                results.extend(batch_results)
            except Exception:
                results.extend([{'original': a, 'error': 'parse_failed'} for a in batch])

        return results

    def standardize_companies(self, company_names: list[str]) -> list[dict]:
        """Standardize company legal forms"""
        batch = company_names[:20]  # Batch
        names_str = "\n".join([f"{i+1}. {n}" for i, n in enumerate(batch)])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": f"""Standardize company names. Extract legal form and clean name.

{names_str}

Return JSON: [{{"original": "...", "legal_form": "ООО|ОАО|ИП|Ltd|Inc|...", "clean_name": "name without legal form", "country": "RU|US|..."}}]"""
            }]
        )

        try:
            import json
            return json.loads(response.content[0].text)
        except Exception:
            return [{'original': n} for n in batch]

Benefits and Results

Problems We Solve

• Transliteration and typos: 'Ivan Ivanov' vs 'Ivan Ivanov' — fuzzy ratio 0.74, but with email/phone bonus the pair becomes a duplicate.
• Different address formats: 'ул. Ленина, д. 5' and 'Lenina 5' — the LLM standardizes to a unified scheme: country, city, street, building.
• Legal form variations: 'ООО Ромашка' and 'Romashka Ltd' — we extract pure names and legal forms.

Comparison of Approaches

Typical Duplicate Scenarios

We use Python + Pandas for processing, rapidfuzz for fast fuzzy matching, Scikit-learn for ML pair classification, and Anthropic Claude 3.5 for LLM conflict resolution. We embed the solution into your MLOps pipeline via Docker or Apache Airflow.

From our practice: a client with 500K CRM records was losing leads due to duplicates. After system implementation, conversion increased by 15%, and processing time dropped from 3 days to 1 hour. Data quality automation with AI reduces manual labor by 80%.

AI deduplication processes 1M records 40 times faster than manual work.

Our Work Process and Deliverables

Our Work Process

1. Data audit — quality analysis, duplicate pattern identification.
2. Rule configuration — thresholds, blocking, LLM prompt tuning.
3. Pipeline development — integration via API or ETL.
4. Documentation and training — process description, Metabase, team training.
5. Post-release support — 2 weeks of monitoring.

What's Included

• Cleaned dataset without duplicates, preserving all unique records.
• Detailed report of found and merged duplicates.
• Documentation on deduplication rules and pipeline configuration.
• API or ETL integration access.
• Training for your team.
• 2 weeks of post-release support.

Timeline and Pricing

Timeline: 2 to 6 weeks depending on data volume and rule complexity. Pricing is determined individually: we assess the number of records, fields, and need for LLM refinements. Request a consultation for your project—free data analysis is included.

Common Pitfalls in Manual Deduplication

• Using only exact matching.
• Ignoring transliteration and typos.
• No normalization before comparison.
• Picking a random canonical record instead of the most complete one.

Fuzzy deduplication of 1M records: 15–30 minutes (depends on average block size). Duplicate detection accuracy at threshold=0.85: precision ~92%, recall ~88%. LLM conflict resolution is applied only to 5–10% of pairs (the rest are merged automatically), keeping AI processing cost manageable.

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.