Data Governance & Compliance

Data Lineage Example:

Source Systems                Transformations              Destinations
┌──────────────┐             ┌────────────────┐           ┌─────────────┐
│  PostgreSQL  │────────────>│  Spark ETL Job │──────────>│  Redshift   │
│  orders      │             │  • Join users  │           │  fact_sales │
│  - order_id  │             │  • Aggregate   │           │  - order_id │
│  - user_id   │             │  • Filter 2024 │           │  - revenue  │
│  - amount    │             └────────────────┘           │  - customer │
└──────────────┘                     │                    └─────────────┘
                                     │                            │
┌──────────────┐                     │                            │
│  MySQL       │─────────────────────┘                            │
│  users       │                                                  │
│  - user_id   │                                                  ▼
│  - name      │                                          ┌───────────────┐
│  - region    │                                          │  Tableau      │
└──────────────┘                                          │  Sales Report │
                                                          └───────────────┘

Lineage Query: "Where does fact_sales.customer come from?"
Answer: PostgreSQL.orders.user_id → joined with MySQL.users.name

Impact Analysis: "What breaks if we delete orders.user_id?"
Answer: Spark ETL job fails, fact_sales.customer becomes NULL, Tableau report breaks

# ============ STEP 1: Configure Atlas for Spark ============
# Add to spark-defaults.conf:
spark.extraListeners=org.apache.atlas.spark.listeners.SparkAtlasEventListener
spark.sql.catalog.implementation=org.apache.atlas.spark.listeners.AtlasCatalog

# Set Atlas connection in atlas-application.properties:
atlas.rest.address=http://atlas-server:21000
atlas.cluster.name=production


# ============ STEP 2: Run Spark Job (Atlas auto-captures lineage) ============
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("AtlasLineageExample") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hive.HiveSessionCatalog") \
    .enableHiveSupport() \
    .getOrCreate()

# Read source tables
orders = spark.read.table("raw.orders")
users = spark.read.table("raw.users")

# Transformation (Atlas automatically tracks lineage)
fact_sales = orders.join(users, "user_id") \
    .groupBy("user_id", users.name.alias("customer")) \
    .agg({"amount": "sum"}) \
    .withColumnRenamed("sum(amount)", "revenue")

# Write to data warehouse (lineage captured)
fact_sales.write.mode("overwrite").saveAsTable("warehouse.fact_sales")

print("✓ Job completed. Atlas captured lineage automatically.")


# ============ STEP 3: Query Lineage via Atlas REST API ============
import requests
import json

atlas_url = "http://atlas-server:21000/api/atlas/v2"
auth = ("admin", "admin")

# Get entity by qualified name
entity_response = requests.get(
    f"{atlas_url}/entity/uniqueAttribute/type/hive_table",
    params={"attr:qualifiedName": "warehouse.fact_sales@production"},
    auth=auth
)
entity = entity_response.json()['entity']
entity_guid = entity['guid']

# Get lineage
lineage_response = requests.get(
    f"{atlas_url}/lineage/{entity_guid}",
    params={"depth": 5, "direction": "BOTH"},
    auth=auth
)
lineage = lineage_response.json()

print("\nLineage for warehouse.fact_sales:")
for relation in lineage['relations']:
    from_entity = relation['fromEntityId']
    to_entity = relation['toEntityId']
    print(f"  {from_entity} → {to_entity}")

"""
Output:
Lineage for warehouse.fact_sales:
  raw.orders@production → warehouse.fact_sales@production
  raw.users@production → warehouse.fact_sales@production
  warehouse.fact_sales@production → tableau.sales_report@production
"""

# Search for all tables containing PII (Personally Identifiable Information)
search_response = requests.get(
    f"{atlas_url}/search/basic",
    params={
        "typeName": "hive_table",
        "classification": "PII",  # Tag applied to tables with sensitive data
        "limit": 100
    },
    auth=auth
)

pii_tables = search_response.json()['entities']

print("Tables containing PII:")
for table in pii_tables:
    print(f"  • {table['attributes']['qualifiedName']}")
    print(f"    Owner: {table['attributes'].get('owner', 'unknown')}")
    print(f"    Classifications: {[c['typeName'] for c in table['classifications']]}")

"""
Output:
Tables containing PII:
  • raw.users@production
    Owner: data-engineering
    Classifications: ['PII', 'GDPR']
  • raw.orders@production
    Owner: data-engineering
    Classifications: ['PII', 'CCPA']
  • warehouse.customer_360@production
    Owner: analytics
    Classifications: ['PII', 'GDPR', 'CCPA']
"""


# Find all downstream dependencies of a table (impact analysis)
def get_downstream_tables(qualified_name, depth=3):
    """Find all tables that depend on this table"""
    entity = requests.get(
        f"{atlas_url}/entity/uniqueAttribute/type/hive_table",
        params={"attr:qualifiedName": qualified_name},
        auth=auth
    ).json()['entity']

    lineage = requests.get(
        f"{atlas_url}/lineage/{entity['guid']}",
        params={"depth": depth, "direction": "OUTPUT"},
        auth=auth
    ).json()

    downstream = set()
    for relation in lineage.get('relations', []):
        downstream.add(relation['toEntityId'])

    return list(downstream)

# What breaks if we delete raw.users?
impact = get_downstream_tables("raw.users@production")
print(f"\nImpact of deleting raw.users: {len(impact)} downstream tables affected")
for table in impact:
    print(f"  ⚠️  {table}")

import boto3
import pandas as pd

glue = boto3.client('glue', region_name='us-east-1')

# ============ Create a database and table in Data Catalog ============
# Create database
glue.create_database(
    DatabaseInput={
        'Name': 'sales_db',
        'Description': 'Sales data warehouse',
        'LocationUri': 's3://my-data-lake/sales/',
        'Parameters': {
            'classification': 'parquet',
            'owner': 'data-engineering',
            'sensitivity': 'internal'
        }
    }
)

# Register table (crawler can do this automatically)
glue.create_table(
    DatabaseName='sales_db',
    TableInput={
        'Name': 'orders',
        'StorageDescriptor': {
            'Columns': [
                {'Name': 'order_id', 'Type': 'bigint'},
                {'Name': 'user_id', 'Type': 'bigint'},
                {'Name': 'amount', 'Type': 'decimal(10,2)'},
                {'Name': 'order_date', 'Type': 'date'}
            ],
            'Location': 's3://my-data-lake/sales/orders/',
            'InputFormat': 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat',
            'OutputFormat': 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat',
            'SerdeInfo': {
                'SerializationLibrary': 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
            }
        },
        'PartitionKeys': [
            {'Name': 'year', 'Type': 'int'},
            {'Name': 'month', 'Type': 'int'}
        ],
        'Parameters': {
            'classification': 'parquet',
            'has_pii': 'false',
            'data_owner': 'sales-team'
        }
    }
)

print("✓ Table registered in Glue Data Catalog")


# ============ Query metadata from Data Catalog ============
# Get all tables in database
response = glue.get_tables(DatabaseName='sales_db')
tables = response['TableList']

print(f"\nTables in sales_db: {len(tables)}")
for table in tables:
    print(f"\n  Table: {table['Name']}")
    print(f"  Location: {table['StorageDescriptor']['Location']}")
    print(f"  Columns: {len(table['StorageDescriptor']['Columns'])}")

    # Show column details
    for col in table['StorageDescriptor']['Columns']:
        print(f"    - {col['Name']}: {col['Type']}")

"""
Output:
✓ Table registered in Glue Data Catalog

Tables in sales_db: 1

  Table: orders
  Location: s3://my-data-lake/sales/orders/
  Columns: 4
    - order_id: bigint
    - user_id: bigint
    - amount: decimal(10,2)
    - order_date: date
"""


# ============ Schema versioning ============
# Update table schema (add new column)
table = glue.get_table(DatabaseName='sales_db', Name='orders')['Table']

# Add new column
table['StorageDescriptor']['Columns'].append({
    'Name': 'payment_method',
    'Type': 'string',
    'Comment': 'Added 2024-01-15'
})

# Update table (creates new version)
glue.update_table(
    DatabaseName='sales_db',
    TableInput={
        'Name': table['Name'],
        'StorageDescriptor': table['StorageDescriptor'],
        'PartitionKeys': table.get('PartitionKeys', []),
        'Parameters': table.get('Parameters', {})
    }
)

# Get version history
versions = glue.get_table_versions(
    DatabaseName='sales_db',
    TableName='orders',
    MaxResults=10
)

print(f"\nSchema versions for orders: {len(versions['TableVersions'])}")
for version in versions['TableVersions']:
    version_id = version['VersionId']
    columns = len(version['Table']['StorageDescriptor']['Columns'])
    print(f"  Version {version_id}: {columns} columns")

"""
Output:
Schema versions for orders: 2
  Version 0: 4 columns
  Version 1: 5 columns (added payment_method)
"""

# Create a crawler to automatically discover schemas in S3
glue.create_crawler(
    Name='sales-crawler',
    Role='arn:aws:iam::123456789012:role/GlueServiceRole',
    DatabaseName='sales_db',
    Targets={
        'S3Targets': [
            {
                'Path': 's3://my-data-lake/sales/',
                'Exclusions': ['**/temp/**', '**/_tmp/**']
            }
        ]
    },
    Schedule='cron(0 2 * * ? *)',  # Run daily at 2 AM
    SchemaChangePolicy={
        'UpdateBehavior': 'UPDATE_IN_DATABASE',  # Update schema if changed
        'DeleteBehavior': 'LOG'  # Log deletions, don't remove from catalog
    },
    Configuration='''{
        "Version": 1.0,
        "CrawlerOutput": {
            "Partitions": {"AddOrUpdateBehavior": "InheritFromTable"}
        }
    }'''
)

# Start the crawler
glue.start_crawler(Name='sales-crawler')
print("✓ Crawler started. Will discover schemas in s3://my-data-lake/sales/")

# Check crawler status
import time
while True:
    crawler = glue.get_crawler(Name='sales-crawler')['Crawler']
    state = crawler['State']
    print(f"  Crawler state: {state}")

    if state == 'READY':
        metrics = crawler.get('LastCrawl', {}).get('Status')
        print(f"  Tables updated: {crawler.get('LastCrawl', {}).get('TablesUpdated', 0)}")
        print(f"  Tables added: {crawler.get('LastCrawl', {}).get('TablesCreated', 0)}")
        break

    time.sleep(10)

"""
Output:
✓ Crawler started. Will discover schemas in s3://my-data-lake/sales/
  Crawler state: RUNNING
  Crawler state: READY
  Tables updated: 2
  Tables added: 3
"""

# Challenge: User requests deletion. Data exists in 200+ tables across:
# - PostgreSQL (OLTP), Redshift (warehouse), S3 (data lake), Elasticsearch, Redis cache

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
import boto3

spark = SparkSession.builder.appName("GDPR-Deletion").getOrCreate()

def delete_user_data(user_id: int):
    """Delete all data for user_id across all systems"""

    print(f"\n🔴 Processing GDPR deletion for user_id={user_id}")

    # ============ STEP 1: Find all tables with user data (using metadata) ============
    glue = boto3.client('glue')
    s3 = boto3.client('s3')

    # Query Glue Data Catalog for tables tagged with PII
    databases = glue.get_databases()['DatabaseList']
    tables_with_pii = []

    for db in databases:
        tables = glue.get_tables(DatabaseName=db['Name'])['TableList']
        for table in tables:
            # Check if table has user_id column and is tagged as PII
            columns = [c['Name'] for c in table['StorageDescriptor']['Columns']]
            has_pii = table.get('Parameters', {}).get('has_pii') == 'true'

            if 'user_id' in columns and has_pii:
                tables_with_pii.append({
                    'database': db['Name'],
                    'table': table['Name'],
                    'location': table['StorageDescriptor']['Location']
                })

    print(f"Found {len(tables_with_pii)} tables with user data")


    # ============ STEP 2: Delete from data lake (S3/Parquet) ============
    for tbl in tables_with_pii:
        print(f"  Processing {tbl['database']}.{tbl['table']}")

        # Read table
        s3_path = tbl['location']
        df = spark.read.parquet(s3_path)

        # Filter out user's data
        df_filtered = df.filter(col("user_id") != user_id)

        # Overwrite table (or write to temp location and swap)
        temp_path = f"{s3_path}_temp"
        df_filtered.write.mode("overwrite").parquet(temp_path)

        # Atomic swap (delete old, rename temp)
        # In production: use Delta Lake MERGE or Iceberg DELETE for ACID
        print(f"    ✓ Removed user data from {tbl['table']}")


    # ============ STEP 3: Delete from data warehouse (Redshift) ============
    import psycopg2

    redshift_conn = psycopg2.connect(
        host='redshift-cluster.us-east-1.redshift.amazonaws.com',
        port=5439,
        dbname='analytics',
        user='admin',
        password='password'
    )
    cursor = redshift_conn.cursor()

    # Find all tables with user_id column
    cursor.execute("""
        SELECT table_schema, table_name
        FROM information_schema.columns
        WHERE column_name = 'user_id'
    """)

    for schema, table in cursor.fetchall():
        cursor.execute(f"DELETE FROM {schema}.{table} WHERE user_id = %s", (user_id,))
        print(f"    ✓ Deleted from Redshift: {schema}.{table}")

    redshift_conn.commit()


    # ============ STEP 4: Delete from Elasticsearch (search index) ============
    from elasticsearch import Elasticsearch

    es = Elasticsearch(['http://elasticsearch:9200'])

    # Delete from all indices
    es.delete_by_query(
        index='users,orders,events',
        body={"query": {"term": {"user_id": user_id}}}
    )
    print(f"    ✓ Deleted from Elasticsearch")


    # ============ STEP 5: Delete from Redis cache ============
    import redis

    redis_client = redis.Redis(host='redis', port=6379)

    # Delete all keys related to user
    keys_to_delete = redis_client.keys(f"user:{user_id}:*")
    if keys_to_delete:
        redis_client.delete(*keys_to_delete)
    print(f"    ✓ Cleared from Redis cache ({len(keys_to_delete)} keys)")


    # ============ STEP 6: Log deletion for audit trail ============
    import datetime

    deletion_record = {
        'user_id': user_id,
        'deleted_at': datetime.datetime.now(datetime.timezone.utc).isoformat(),
        'tables_affected': len(tables_with_pii),
        'status': 'completed'
    }

    # Store in append-only audit log (for compliance proof)
    audit_log = spark.createDataFrame([deletion_record])
    audit_log.write.mode("append").parquet("s3://audit-logs/gdpr-deletions/")

    print(f"\n✅ GDPR deletion completed for user_id={user_id}")
    print(f"   Tables affected: {len(tables_with_pii)}")
    print(f"   Audit log: s3://audit-logs/gdpr-deletions/")

    return deletion_record


# Execute deletion
result = delete_user_data(user_id=12345)

"""
Output:
🔴 Processing GDPR deletion for user_id=12345
Found 47 tables with user data
  Processing sales_db.orders
    ✓ Removed user data from orders
  Processing sales_db.events
    ✓ Removed user data from events
  [... 45 more tables ...]
    ✓ Deleted from Redshift: public.fact_orders
    ✓ Deleted from Redshift: analytics.user_metrics
    ✓ Deleted from Elasticsearch
    ✓ Cleared from Redis cache (15 keys)

✅ GDPR deletion completed for user_id=12345
   Tables affected: 47
   Audit log: s3://audit-logs/gdpr-deletions/

Execution time: 4.2 seconds
"""

# User requests all their personal data (GDPR "Right to Access")

def export_user_data(user_id: int, output_format='json'):
    """Export all user data across systems in machine-readable format"""

    print(f"📦 Exporting all data for user_id={user_id}")

    user_data = {}

    # ============ STEP 1: Collect from all sources ============
    # Data lake (S3)
    for tbl in tables_with_pii:
        df = spark.read.parquet(tbl['location'])
        user_rows = df.filter(col("user_id") == user_id).toPandas()

        if not user_rows.empty:
            user_data[f"{tbl['database']}.{tbl['table']}"] = user_rows.to_dict('records')

    # Redshift
    cursor.execute("SELECT * FROM public.users WHERE user_id = %s", (user_id,))
    user_data['redshift.users'] = [dict(zip([d[0] for d in cursor.description], row))
                                    for row in cursor.fetchall()]

    # Elasticsearch
    search_results = es.search(
        index='*',
        body={"query": {"term": {"user_id": user_id}}},
        size=10000
    )
    user_data['elasticsearch'] = [hit['_source'] for hit in search_results['hits']['hits']]


    # ============ STEP 2: Export in requested format ============
    import json

    output_path = f"s3://gdpr-exports/user_{user_id}_export.json"

    with open(f"/tmp/user_{user_id}_export.json", 'w') as f:
        json.dump(user_data, f, indent=2, default=str)

    # Upload to S3 with presigned URL (expires in 7 days)
    s3.upload_file(f"/tmp/user_{user_id}_export.json", 'gdpr-exports', f'user_{user_id}_export.json')

    presigned_url = s3.generate_presigned_url(
        'get_object',
        Params={'Bucket': 'gdpr-exports', 'Key': f'user_{user_id}_export.json'},
        ExpiresIn=604800  # 7 days
    )

    print(f"✅ Data export completed")
    print(f"   Records collected: {sum(len(v) if isinstance(v, list) else 1 for v in user_data.values())}")
    print(f"   Download URL (expires in 7 days):")
    print(f"   {presigned_url}")

    return presigned_url

# Execute export
export_user_data(user_id=12345)

"""
Output:
📦 Exporting all data for user_id=12345
✅ Data export completed
   Records collected: 1,247
   Download URL (expires in 7 days):
   https://gdpr-exports.s3.amazonaws.com/user_12345_export.json?X-Amz-Expires=604800&...
"""

Problem: Customer data scattered across systems with inconsistencies

CRM System:
┌─────────┬───────────────┬──────────────────────┐
│ id      │ name          │ email                │
├─────────┼───────────────┼──────────────────────┤
│ CRM-123 │ John Smith    │ john.smith@email.com │
└─────────┴───────────────┴──────────────────────┘

E-commerce System:
┌─────────┬──────────────┬─────────────────────┐
│ id      │ name         │ email               │
├─────────┼──────────────┼─────────────────────┤
│ WEB-456 │ J. Smith     │ jsmith@email.com    │
└─────────┴──────────────┴─────────────────────┘

Support System:
┌──────────┬───────────────┬──────────────────────┐
│ id       │ name          │ email                │
├──────────┼───────────────┼──────────────────────┤
│ SUP-7890 │ Jonathan S.   │ john.smith@email.com │
└──────────┴───────────────┴──────────────────────┘

❌ Same customer with 3 different IDs, names, and emails!


MDM Solution: Create golden record

Master Data Management System:
┌─────────────────────────────────────────────────────────┐
│  Golden Record (MDM-45678)                              │
│  ┌────────────────────────────────────────────────────┐ │
│  │ Canonical Data:                                    │ │
│  │  • Name: John Smith                                │ │
│  │  │ Email: john.smith@email.com                     │ │
│  │  • Phone: +1-555-0123                              │ │
│  │  • Address: 123 Main St, New York, NY 10001        │ │
│  │                                                    │ │
│  │ Source Systems:                                    │ │
│  │  • CRM: CRM-123                                    │ │
│  │  • E-commerce: WEB-456                             │ │
│  │  • Support: SUP-7890                               │ │
│  │                                                    │ │
│  │ Confidence Score: 95% (based on matching algo)     │ │
│  │ Last Updated: 2024-01-15 10:30:00                  │ │
│  └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

✅ Single source of truth for customer across all systems

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lower, trim, regexp_replace, soundex, levenshtein
from pyspark.ml.feature import HashingTF, MinHashLSH

spark = SparkSession.builder.appName("MDM").getOrCreate()

# ============ STEP 1: Load data from multiple sources ============
crm_customers = spark.read.parquet("s3://data-lake/crm/customers/")
ecom_customers = spark.read.parquet("s3://data-lake/ecommerce/users/")
support_customers = spark.read.parquet("s3://data-lake/support/tickets/")

# Add source system identifier
crm_customers = crm_customers.withColumn("source_system", lit("CRM"))
ecom_customers = ecom_customers.withColumn("source_system", lit("ECOM"))
support_customers = support_customers.withColumn("source_system", lit("SUPPORT"))


# ============ STEP 2: Standardize and clean data ============
def standardize_customer(df):
    """Apply data quality rules"""
    return df.select(
        col("id").alias("source_id"),
        col("source_system"),
        # Standardize name: lowercase, remove extra spaces
        trim(lower(regexp_replace(col("name"), "\\s+", " "))).alias("name_clean"),
        # Standardize email
        lower(trim(col("email"))).alias("email_clean"),
        # Phone: remove non-digits
        regexp_replace(col("phone"), "[^0-9]", "").alias("phone_clean")
    )

crm_clean = standardize_customer(crm_customers)
ecom_clean = standardize_customer(ecom_customers)
support_clean = standardize_customer(support_customers)

# Union all sources
all_customers = crm_clean.union(ecom_clean).union(support_clean)


# ============ STEP 3: Find duplicates using fuzzy matching ============
# Method 1: Exact email match
email_matches = all_customers.alias("a").join(
    all_customers.alias("b"),
    (col("a.email_clean") == col("b.email_clean")) &
    (col("a.source_id") != col("b.source_id")),
    "inner"
).select(
    col("a.source_id").alias("id1"),
    col("b.source_id").alias("id2"),
    lit("email_match").alias("match_type"),
    lit(1.0).alias("confidence")
)

# Method 2: Fuzzy name + phone match
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType

@udf(FloatType())
def name_similarity(name1, name2):
    """Calculate similarity between names (Levenshtein distance)"""
    if not name1 or not name2:
        return 0.0
    max_len = max(len(name1), len(name2))
    distance = levenshtein(name1, name2)
    return 1.0 - (distance / max_len)

name_matches = all_customers.alias("a").join(
    all_customers.alias("b"),
    col("a.phone_clean") == col("b.phone_clean"),
    "inner"
).filter(
    (col("a.source_id") != col("b.source_id")) &
    (name_similarity(col("a.name_clean"), col("b.name_clean")) > 0.8)
).select(
    col("a.source_id").alias("id1"),
    col("b.source_id").alias("id2"),
    lit("name_phone_match").alias("match_type"),
    name_similarity(col("a.name_clean"), col("b.name_clean")).alias("confidence")
)


# Combine all matches
all_matches = email_matches.union(name_matches)


# ============ STEP 4: Create golden records ============
# Group matched records into clusters (connected components)
from graphframes import GraphFrame

# Create graph of matches
vertices = all_customers.select("source_id").distinct().withColumnRenamed("source_id", "id")
edges = all_matches.select(
    col("id1").alias("src"),
    col("id2").alias("dst"),
    col("confidence")
)

graph = GraphFrame(vertices, edges)

# Find connected components (each component = one golden record)
clusters = graph.connectedComponents()

print(f"\nMDM Results:")
print(f"  Total records: {all_customers.count()}")
print(f"  Unique entities (golden records): {clusters.select('component').distinct().count()}")
print(f"  Duplicate rate: {(1 - clusters.select('component').distinct().count() / all_customers.count()) * 100:.1f}%")


# ============ STEP 5: Create master data table ============
# For each cluster, select best values (data quality-based)
def create_golden_record(cluster_id, records):
    """Create canonical record from matched records"""

    # Select most complete email (prefer verified)
    email = records.filter(col("email_clean").isNotNull()) \
                  .orderBy(col("source_system") == "CRM", "desc") \
                  .first()["email_clean"]

    # Select most complete name (longest non-null)
    name = records.filter(col("name_clean").isNotNull()) \
                 .orderBy(length(col("name_clean")).desc()) \
                 .first()["name_clean"]

    # Phone: prefer CRM system
    phone = records.filter(col("phone_clean").isNotNull()) \
                  .orderBy(col("source_system") == "CRM", "desc") \
                  .first()["phone_clean"]

    return {
        'golden_id': f"MDM-{cluster_id}",
        'name': name,
        'email': email,
        'phone': phone,
        'source_systems': [r["source_system"] for r in records.collect()],
        'source_ids': [r["source_id"] for r in records.collect()],
        'created_at': datetime.utcnow().isoformat()
    }

# Apply to all clusters
golden_records = clusters.groupBy("component").apply(create_golden_record)

# Save master data
golden_records.write.mode("overwrite").parquet("s3://data-lake/mdm/golden_records/")

print("\n✅ Golden records created and saved to s3://data-lake/mdm/golden_records/")


# ============ Example output ============
golden_records.show(5, truncate=False)

"""
Output:
MDM Results:
  Total records: 15,234
  Unique entities (golden records): 12,891
  Duplicate rate: 15.4%

✅ Golden records created and saved to s3://data-lake/mdm/golden_records/

+------------+-------------------+-------------------------+--------------+-------------------------+
|golden_id   |name               |email                    |phone         |source_systems           |
+------------+-------------------+-------------------------+--------------+-------------------------+
|MDM-1       |john smith         |john.smith@email.com     |5555550123    |[CRM, ECOM, SUPPORT]     |
|MDM-2       |jane doe           |jane.doe@company.com     |5555550456    |[CRM, ECOM]              |
|MDM-3       |robert johnson     |rjohnson@email.com       |5555550789    |[SUPPORT]                |
|MDM-4       |mary williams      |mary.w@email.com         |5555551234    |[CRM, ECOM]              |
|MDM-5       |james brown        |jbrown@company.com       |5555555678    |[CRM]                    |
+------------+-------------------+-------------------------+--------------+-------------------------+
"""