Home > Notebooks > Silver Layer¶

# Standard Silver cleansing pattern
from pyspark.sql.functions import sha2, regexp_replace, when, col

df_cleansed = (spark.table("bronze.player_profile")
    # Deduplicate by business key
    .dropDuplicates(["player_id", "_ingested_at"])

    # Hash SSN for privacy
    .withColumn("ssn_hash", sha2(col("ssn"), 256))
    .drop("ssn")

    # Mask card numbers (keep last 4)
    .withColumn("card_masked", regexp_replace(col("card_number"), r"\d(?=\d{4})", "*"))

    # Validate and default null tiers
    .withColumn("player_tier", when(col("player_tier").isNull(), "BRONZE").otherwise(col("player_tier")))

    # Add processing metadata
    .withColumn("_processed_at", current_timestamp())
    .withColumn("_is_valid", lit(True))
)

df_cleansed.write.format("delta").mode("overwrite").saveAsTable("silver.player_master")

# Check for duplicates
slot_count = spark.sql("SELECT COUNT(*) FROM silver.slot_cleansed").first()[0]
distinct_count = spark.sql("""
    SELECT COUNT(*) FROM (
        SELECT DISTINCT machine_id, spin_timestamp 
        FROM silver.slot_cleansed
    )
""").first()[0]
assert slot_count == distinct_count, "Duplicates detected!"

# Validate PII masking
pii_exposed = spark.sql("""
    SELECT COUNT(*) 
    FROM silver.player_master 
    WHERE ssn IS NOT NULL OR LENGTH(card_number) = 16
""").first()[0]
assert pii_exposed == 0, "PII not properly masked!"

# Check referential integrity
orphans = spark.sql("""
    SELECT COUNT(*) 
    FROM silver.financial_reconciled f
    LEFT JOIN silver.player_master p ON f.player_id = p.player_id
    WHERE p.player_id IS NULL
""").first()[0]
assert orphans == 0, f"Found {orphans} orphan records!"

# Run Great Expectations checkpoint
import great_expectations as gx

context = gx.get_context()
result = context.run_checkpoint(checkpoint_name="silver_checkpoint")
assert result.success, "Data quality checks failed!"

# Track player tier changes over time
player_history = spark.sql("""
    SELECT 
        player_id,
        player_tier,
        effective_from,
        effective_to,
        is_current
    FROM silver.player_master_history
    WHERE player_id = 'PLY-12345'
    ORDER BY effective_from
""")