Enterprise Data Deduplication Pipeline

SQL-Based Deterministic Matching Example:

WITH normalized_data AS (
    SELECT 
        customer_id,
        LOWER(TRIM(REGEXP_REPLACE(email, '\s+', ''))) AS clean_email,
        REGEXP_REPLACE(phone, '[^0-9]', '') AS clean_phone,
        UPPER(TRIM(first_name)) AS first_name,
        UPPER(TRIM(last_name)) AS last_name,
        source_system
    FROM raw_customers
),
exact_matches AS (
    SELECT 
        a.customer_id AS customer_id_1,
        b.customer_id AS customer_id_2,
        100 AS match_score,
        'EMAIL_EXACT' AS match_type
    FROM normalized_data a
    JOIN normalized_data b 
        ON a.clean_email = b.clean_email
        AND a.customer_id < b.customer_id
    WHERE a.clean_email IS NOT NULL
    
    UNION ALL
    
    SELECT 
        a.customer_id AS customer_id_1,
        b.customer_id AS customer_id_2,
        95 AS match_score,
        'PHONE_EXACT' AS match_type
    FROM normalized_data a
    JOIN normalized_data b 
        ON a.clean_phone = b.clean_phone
        AND a.customer_id < b.customer_id
    WHERE a.clean_phone IS NOT NULL
        AND LENGTH(a.clean_phone) = 10
)
SELECT * FROM exact_matches;

PySpark ETL Job Structure:

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from awsglue.context import GlueContext
from awsglue.job import Job

# Initialize Glue context
glueContext = GlueContext(SparkSession.builder.getOrCreate())
spark = glueContext.spark_session
job = Job(glueContext)

# Read from multiple sources
customers_kay = glueContext.create_dynamic_frame.from_catalog(
    database="signet_raw",
    table_name="kay_customers"
)

customers_zales = glueContext.create_dynamic_frame.from_catalog(
    database="signet_raw",
    table_name="zales_customers"
)

# Data normalization and cleansing
df_normalized = (customers_kay.toDF()
    .withColumn("email_clean", lower(trim(col("email"))))
    .withColumn("phone_clean", regexp_replace(col("phone"), "[^0-9]", ""))
    .withColumn("name_soundex", soundex(col("last_name")))
    .withColumn("source_system", lit("KAY"))
)

# Write to Redshift staging
glueContext.write_dynamic_frame.from_jdbc_conf(
    frame=DynamicFrame.fromDF(df_normalized, glueContext, "normalized"),
    catalog_connection="redshift-connection",
    connection_options={
        "dbtable": "staging.customers_normalized",
        "database": "signet_mdm"
    }
)

job.commit()

AWS AppFlow Configuration (near-real-time):

• Source: Amazon Redshift (deduplicated_customers table)
• Destination: Salesforce (Contact & Account objects)
• Trigger: On-demand and scheduled (every 2 hours)
• Field Mappings: Custom mapping with transformation rules
• Error Handling: CloudWatch logs + SNS notifications

Bulk API 2.0 for High-Volume Updates:

• Batch Size: 10,000 records per batch
• Operation: Upsert using External ID
• Parallel Processing: 5 concurrent jobs
• Monitoring: Job status tracking + retry logic
• Validation: Success/error reports with detailed logging