Entity Match Platform

A production-ready data pipeline for entity matching, web data enrichment, and analytics visualization using Apache Airflow, PySpark, and Docker.

🎯 Overview

This platform enables organizations to:

• Match entities across different data sources using fuzzy matching algorithms
• Enrich data by scraping external sources for additional information
• Visualize results through an interactive analytics dashboard
• Track matching quality with comprehensive metrics and tracing

Perfect for:

• Customer data deduplication
• Third-party data integration
• Entity resolution across databases
• Data quality improvement initiatives
• Master data management (MDM)

🏗️ Architecture

┌─────────────────┐
│   Source A      │ (Internal Database)
│  (Primary Data) │
└────────┬────────┘
         │
         ├──────────► ┌─────────────────────┐
         │            │  Data Preparation   │
         │            │  - Normalization    │
┌────────▼────────┐  │  - Deduplication    │
│   Source B      │──┤  - Web Enrichment   │
│ (External Data) │  └──────────┬──────────┘
└─────────────────┘             │
                                │
                    ┌───────────▼──────────┐
                    │   Entity Matching    │
                    │  - Exact Matching    │
                    │  - LSH Fuzzy Match   │
                    │  - Multi-stage       │
                    └───────────┬──────────┘
                                │
                    ┌───────────▼──────────┐
                    │    Visualization     │
                    │  - Match Analytics   │
                    │  - Quality Metrics   │
                    └──────────────────────┘

✨ Features

1. Multi-Stage Matching Engine

• Exact Matching: Fast matching on unique identifiers (tax IDs, registration numbers)
• Fuzzy Matching: LSH-based similarity matching for name fields
• Weighted Scoring: Configurable weights for different field importance
• Deduplication: Automatic removal of duplicate matches

2. Web Data Enrichment

• Configurable web scraping for external data sources
• Intelligent search strategies (by ID, name, etc.)
• Automatic data validation and cleaning
• Rate limiting and error handling

3. Production-Ready Pipeline

• Docker containerization for consistent deployment
• Apache Airflow for workflow orchestration
• PySpark for distributed data processing
• Incremental processing to avoid reprocessing
• Tracing & auditing of all matching attempts

4. Interactive Dashboard

• Real-time matching statistics
• Data quality metrics
• Field-level mismatch analysis
• Tracing analytics with attempt history

🚀 Quick Start

Prerequisites

• Docker & Docker Compose
• 8GB+ RAM recommended
• Python 3.9+

Installation

1. Clone the repository

git clone https://github.com/Chaimaaorg/entity-match-platform.git
cd entity-match-platform

2. Configure environment

cp .env.example .env
# Edit .env with your settings

3. Update configuration Edit config/config.ini:

[dev]
db = local_db
source_a_main = /app/data/source_a/entities.csv
source_b_main = /app/data/source_b/entities.csv

4. Start the platform

docker-compose up -d

5. Access services

• Airflow UI: http://localhost:8080 (user: admin, password: admin)
• Dashboard: Open visualization/analytics_dashboard.html in browser

Running the Pipeline

1. Prepare your data

   • Place Source A data in data/source_a/
   • Place Source B data in data/source_b/
   • Ensure CSV files have proper headers

2. Trigger the DAG

   • Open Airflow UI
   • Enable the entity_matching_pipeline DAG
   • Trigger manually or wait for scheduled run

3. View results

   • Matched records: data/matched/results.csv
   • Processing logs: Check Airflow task logs
   • Analytics: Load CSV files into the dashboard

📊 Data Format

Source A (Primary Dataset)

entity_id,name,tax_id,registration_number,city
001,Acme Corp,TX123456,REG789,New York
002,Beta LLC,TX234567,REG890,Los Angeles

Source B (Secondary Dataset)

entity_id,name,tax_id,registration_number,activity
B001,Acme Corporation,TX123456,REG789,Manufacturing
B002,Beta Limited,TX234567,REG890,Retail

Required Fields

• Unique identifier (entity_id, company_id, etc.)
• Name field (company name, organization name, etc.)
• Optional identifiers (tax_id, registration_number, etc.)

⚙️ Configuration

Matching Parameters (src/matching/matcher.py)

@dataclass
class MatchingConfig:
    lsh_num_hash_tables: int = 3
    lsh_distance_threshold: float = 0.5
    similarity_threshold: float = 0.65
    max_candidates_per_record: int = 10
    
    weights: Dict[str, float] = {
        "name": 0.45,
        "registration_id": 0.25,
        "tax_id": 0.25,
        "other": 0.05
    }

Web Enrichment (src/enrichment/web_scraper.py)

enricher = EntityEnricher(
    base_url="https://your-data-source.com/search",
    search_params_mapping={
        "tax_id": "tax_param",
        "registration_id": "reg_param",
        "name": "name_param"
    }
)

📁 Project Structure

entity-match-platform/
├── airflow/                 # Airflow DAGs and configuration
│   ├── dags/
│   │   └── entity_matching_pipeline.py
│   └── Dockerfile
├── config/                  # Configuration files
│   └── config.ini
├── src/                     # Source code
│   ├── enrichment/          # Web scraping modules
│   │   └── web_scraper.py
│   └── matching/            # Matching engine
│       ├── data_preparation.py
│       ├── matcher.py
│       └── utils.py
├── data/                    # Data storage
│   ├── source_a/            # Primary dataset
│   ├── source_b/            # Secondary dataset
│   ├── processed/           # Cleaned data
│   └── matched/             # Matching results
├── visualization/           # Analytics dashboard
│   └── analytics_dashboard.html
└── tests/                   # Unit tests

🔧 Customization

Adding New Data Sources

1. Update config/config.ini with new paths
2. Create data loaders in src/matching/data_preparation.py
3. Adjust field mappings in matching configuration

Custom Matching Logic

Override methods in src/matching/matcher.py:

class CustomMatcher(LSHMatcher):
    def _build_feature_pipeline(self, company_col: str):
        # Custom text processing logic
        pass

Web Scraping Configuration

Implement scrape_entities_from_html() for your target website:

def scrape_entities_from_html(self, html: str):
    # Parse HTML specific to your data source
    soup = BeautifulSoup(html, "html.parser")
    # Extract entity information
    return entities

🤝 Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Submit a pull request

📄 License

MIT License - see LICENSE file for details