Deduplication in Machine Learning

import hashlib
from collections import defaultdict
from typing import List, Dict, Tuple

def exact_dedup(documents: List[str]) -> Tuple[List[str], Dict[str, List[int]]]:
    """Remove exact duplicates using MD5 hashing.
    
    Returns:
        - Deduplicated list of documents
        - Mapping of hash -> list of original indices (for audit trail)
    """
    hash_to_indices = defaultdict(list)
    
    for idx, doc in enumerate(documents):
        # Normalize whitespace before hashing
        normalized = " ".join(doc.split()).strip().lower()
        doc_hash = hashlib.md5(normalized.encode("utf-8")).hexdigest()
        hash_to_indices[doc_hash].append(idx)
    
    # Keep the first occurrence of each unique document
    deduplicated = []
    duplicate_clusters = {}
    
    for doc_hash, indices in hash_to_indices.items():
        deduplicated.append(documents[indices[0]])
        if len(indices) > 1:
            duplicate_clusters[doc_hash] = indices
    
    n_duplicates = len(documents) - len(deduplicated)
    print(f"Removed {n_duplicates} exact duplicates "
          f"({100 * n_duplicates / len(documents):.1f}% of dataset)")
    
    return deduplicated, duplicate_clusters

# Example usage
documents = [
    "The quick brown fox jumps over the lazy dog.",
    "Machine learning is transforming industries worldwide.",
    "The quick brown fox jumps over the lazy dog.",  # exact dup
    "  The quick brown   fox jumps over the lazy dog.  ",  # whitespace variant
    "A completely different document about data science.",
]

deduped, clusters = exact_dedup(documents)
# Output: Removed 2 exact duplicates (40.0% of dataset)

from datasketch import MinHash, MinHashLSH
from typing import List, Set, Tuple
import re

def create_shingles(text: str, k: int = 5) -> Set[str]:
    """Convert text to a set of character-level k-shingles."""
    text = re.sub(r"\s+", " ", text.lower().strip())
    return {text[i:i+k] for i in range(len(text) - k + 1)}

def build_minhash(shingles: Set[str], num_perm: int = 128) -> MinHash:
    """Build a MinHash signature from a shingle set."""
    m = MinHash(num_perm=num_perm)
    for shingle in shingles:
        m.update(shingle.encode("utf-8"))
    return m

def fuzzy_dedup(
    documents: List[str],
    threshold: float = 0.8,
    num_perm: int = 128,
    shingle_size: int = 5,
) -> Tuple[List[str], List[Set[int]]]:
    """Deduplicate documents using MinHash LSH.
    
    Args:
        documents: List of document strings
        threshold: Jaccard similarity threshold for duplicates
        num_perm: Number of MinHash permutations (higher = more accurate)
        shingle_size: Character n-gram size for shingling
    
    Returns:
        - Deduplicated documents
        - List of duplicate clusters (sets of original indices)
    """
    # Initialize LSH index
    lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
    minhashes = {}
    
    # Build signatures and insert into LSH
    for idx, doc in enumerate(documents):
        shingles = create_shingles(doc, k=shingle_size)
        if len(shingles) == 0:
            continue
        mh = build_minhash(shingles, num_perm=num_perm)
        minhashes[idx] = mh
        try:
            lsh.insert(str(idx), mh)
        except ValueError:
            pass  # Duplicate MinHash signature already in index
    
    # Find duplicate clusters using Union-Find
    parent = list(range(len(documents)))
    
    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]  # path compression
            x = parent[x]
        return x
    
    def union(x, y):
        px, py = find(x), find(y)
        if px != py:
            parent[px] = py
    
    # Query each document against LSH index
    for idx in minhashes:
        candidates = lsh.query(minhashes[idx])
        for cand in candidates:
            cand_idx = int(cand)
            if cand_idx != idx:
                # Verify with actual Jaccard estimate
                sim = minhashes[idx].jaccard(minhashes[cand_idx])
                if sim >= threshold:
                    union(idx, cand_idx)
    
    # Build clusters and select representatives
    from collections import defaultdict
    clusters = defaultdict(set)
    for idx in range(len(documents)):
        clusters[find(idx)].add(idx)
    
    # Keep the first document in each cluster
    deduplicated = []
    duplicate_groups = []
    for root, members in clusters.items():
        representative = min(members)  # keep earliest
        deduplicated.append(documents[representative])
        if len(members) > 1:
            duplicate_groups.append(members)
    
    n_removed = len(documents) - len(deduplicated)
    print(f"Removed {n_removed} near-duplicates "
          f"({100 * n_removed / len(documents):.1f}% of dataset)")
    
    return deduplicated, duplicate_groups

import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.cluster import MiniBatchKMeans
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Tuple, Set
from collections import defaultdict

def semantic_dedup(
    documents: List[str],
    model_name: str = "all-MiniLM-L6-v2",
    n_clusters: int = 100,
    similarity_threshold: float = 0.9,
    batch_size: int = 256,
) -> Tuple[List[str], List[Set[int]]]:
    """Deduplicate using embedding-based semantic similarity.
    
    Uses SemDeDup approach: cluster embeddings, then deduplicate
    within clusters to avoid O(n^2) pairwise comparison.
    
    Args:
        documents: List of document strings
        model_name: Sentence transformer model name
        n_clusters: Number of k-means clusters (controls granularity)
        similarity_threshold: Cosine similarity threshold for duplicates
        batch_size: Encoding batch size
    """
    # Step 1: Generate embeddings
    model = SentenceTransformer(model_name)
    embeddings = model.encode(
        documents, batch_size=batch_size, show_progress_bar=True,
        normalize_embeddings=True  # for cosine similarity
    )
    
    # Step 2: Cluster embeddings to reduce comparison space
    kmeans = MiniBatchKMeans(
        n_clusters=min(n_clusters, len(documents) // 2),
        batch_size=1024, random_state=42
    )
    cluster_labels = kmeans.fit_predict(embeddings)
    
    # Step 3: Deduplicate within each cluster
    parent = list(range(len(documents)))
    
    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x
    
    def union(x, y):
        px, py = find(x), find(y)
        if px != py:
            parent[px] = py
    
    cluster_to_docs = defaultdict(list)
    for idx, label in enumerate(cluster_labels):
        cluster_to_docs[label].append(idx)
    
    for label, doc_indices in cluster_to_docs.items():
        if len(doc_indices) < 2:
            continue
        cluster_embeds = embeddings[doc_indices]
        sim_matrix = cosine_similarity(cluster_embeds)
        
        for i in range(len(doc_indices)):
            for j in range(i + 1, len(doc_indices)):
                if sim_matrix[i, j] >= similarity_threshold:
                    union(doc_indices[i], doc_indices[j])
    
    # Step 4: Build clusters and select representatives
    clusters = defaultdict(set)
    for idx in range(len(documents)):
        clusters[find(idx)].add(idx)
    
    deduplicated = []
    duplicate_groups = []
    for root, members in clusters.items():
        representative = min(members)
        deduplicated.append(documents[representative])
        if len(members) > 1:
            duplicate_groups.append(members)
    
    n_removed = len(documents) - len(deduplicated)
    print(f"Removed {n_removed} semantic duplicates "
          f"({100 * n_removed / len(documents):.1f}% of dataset)")
    
    return deduplicated, duplicate_groups

import dedupe
import csv
from collections import OrderedDict

def load_data(filepath: str) -> dict:
    """Load structured records for deduplication."""
    data = OrderedDict()
    with open(filepath, "r") as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader):
            data[i] = {k: v.strip() if v else None for k, v in row.items()}
    return data

def deduplicate_records(data: dict) -> list:
    """Deduplicate structured records using the Dedupe library.
    
    Uses active learning to train a matching model.
    """
    # Define fields and their types for comparison
    fields = [
        {"field": "name", "type": "String", "has missing": True},
        {"field": "address", "type": "String", "has missing": True},
        {"field": "city", "type": "ShortString", "has missing": True},
        {"field": "phone", "type": "String", "has missing": True},
        {"field": "email", "type": "String", "has missing": True},
    ]
    
    # Initialize deduper
    deduper = dedupe.Dedupe(fields)
    
    # Prepare training data (sample pairs for active learning)
    deduper.prepare_training(data, sample_size=15000)
    
    # In production, you would use:
    # dedupe.console_label(deduper)  # Interactive labeling
    # Or load pre-labeled training data:
    # with open("training.json", "r") as f:
    #     deduper.prepare_training(data, training_file=f)
    
    # Train the matching model
    deduper.train()
    
    # Find duplicate clusters
    # threshold: probability threshold for considering a match
    clustered_dupes = deduper.partition(data, threshold=0.5)
    
    # Process results
    cluster_membership = {}
    for cluster_id, (records, scores) in enumerate(clustered_dupes):
        for record_id, score in zip(records, scores):
            cluster_membership[record_id] = {
                "cluster_id": cluster_id,
                "confidence": score,
            }
    
    n_clusters = len(clustered_dupes)
    n_duplicates = len(data) - n_clusters
    print(f"Found {n_clusters} unique entities from {len(data)} records")
    print(f"Identified {n_duplicates} duplicate records")
    
    return clustered_dupes

# Deduplication pipeline configuration (YAML)
pipeline:
  name: training-data-dedup
  stages:
    - name: exact_dedup
      method: md5
      normalize:
        lowercase: true
        strip_whitespace: true
        unicode_nfkc: true

    - name: fuzzy_dedup
      method: minhash_lsh
      params:
        num_permutations: 128
        jaccard_threshold: 0.8
        shingle_size: 5
        bands: 20
        rows_per_band: 5
      backend: spark  # or 'ray', 'dask', 'gpu'

    - name: semantic_dedup
      enabled: false  # enable for high-value datasets
      method: semdedup
      params:
        model: all-MiniLM-L6-v2
        n_clusters: 1000
        cosine_threshold: 0.92

  output:
    deduplicated_path: s3://data-lake/deduped/
    duplicate_log_path: s3://data-lake/dedup-audit/
    format: parquet
    compression: snappy