🥈 Tutorial 02: Silver Layer¶

# Cell 1: Configuration (imports + parameter shim + source/target paths)
#
# The parameter shim (`_get_arg`) is defined in the SAME cell as its caller to
# avoid NameError when cells are run out of order after importing the notebook
# from the repo. Keep imports, the shim, and the batch_id assignment together.
import os
from datetime import datetime

from delta.tables import DeltaTable
from pyspark.sql.functions import (
    array, array_compact, col, count, current_timestamp, filter,
    hour, initcap, lit, to_date, to_timestamp, trim, upper, when,
)
from pyspark.sql.types import (
    DateType, DecimalType, IntegerType, StringType,
    StructField, StructType, TimestampType,
)


def _get_arg(name: str, default=None):
    """Read a parameter from Fabric/Synapse runtime, env var, or default."""
    try:
        import notebookutils  # Fabric runtime
        return notebookutils.notebook.getArgument(name, default)
    except Exception:
        try:
            import mssparkutils  # legacy Synapse/Fabric runtime
            return mssparkutils.notebook.getArgument(name, default)
        except Exception:
            return os.environ.get(name.upper(), default)


# Parameters (set by pipeline or manual)
batch_id = _get_arg("batch_id", datetime.now().strftime("%Y%m%d_%H%M%S"))

# Source and target — three-part names require schema-enabled Lakehouses
source_table = "lh_bronze.dbo.bronze_slot_telemetry"
SILVER_TABLE = "lh_silver.dbo.silver_slot_cleansed"

print(f"Processing batch: {batch_id}")
print(f"Source: {source_table}")
print(f"Target: {SILVER_TABLE}")

# Cell 2: Read Bronze Data
df_bronze = spark.table(source_table)

print(f"\n📊 Bronze Layer Statistics:")
print(f"  Records: {df_bronze.count():,}")
print(f"  Columns: {len(df_bronze.columns)}")
df_bronze.printSchema()

# Cell 3: Data Quality Checks
print("\n🔍 Data Quality Analysis:")
print("-" * 40)

# Rule 1: event_id not null and unique
null_event_ids = df_bronze.filter(col("event_id").isNull()).count()
print(f"❌ Null event_ids: {null_event_ids}")

# Rule 2: Valid event types
valid_events = [
    "GAME_PLAY", "JACKPOT", "METER_UPDATE", "DOOR_OPEN", "DOOR_CLOSE",
    "BILL_ACCEPTED", "TICKET_PRINTED", "HAND_PAY", "TILT", "POWER_OFF", "POWER_ON"
]
invalid_events = df_bronze.filter(~col("event_type").isin(valid_events)).count()
print(f"❌ Invalid event types: {invalid_events}")

# Rule 3: coin_in >= 0 when present
negative_coin_in = df_bronze.filter(col("coin_in") < 0).count()
print(f"❌ Negative coin_in: {negative_coin_in}")

# Rule 4: event_timestamp within expected range
future_events = df_bronze.filter(col("event_timestamp") > current_timestamp()).count()
print(f"❌ Future timestamps: {future_events}")

# Rule 5: Valid denomination values
valid_denoms = [0.01, 0.05, 0.25, 0.50, 1.00, 5.00, 25.00, 100.00]
invalid_denoms = df_bronze.filter(
    col("denomination").isNotNull() & ~col("denomination").isin(valid_denoms)
).count()
print(f"❌ Invalid denominations: {invalid_denoms}")

# Summary
total_issues = null_event_ids + invalid_events + negative_coin_in + future_events
total_records = df_bronze.count()
quality_score = ((total_records - total_issues) / total_records) * 100 if total_records > 0 else 0
print("-" * 40)
print(f"📈 Initial Quality Score: {quality_score:.2f}%")

# Cell 4: Apply Cleansing Transformations
print("\n🧹 Applying Cleansing Transformations...")

df_cleansed = df_bronze \
    .filter(col("event_id").isNotNull()) \
    .filter(col("event_type").isin(valid_events)) \
    .filter(col("event_timestamp") <= current_timestamp()) \
    .withColumn("coin_in", when(col("coin_in") < 0, 0).otherwise(col("coin_in"))) \
    .withColumn("coin_out", when(col("coin_out") < 0, 0).otherwise(col("coin_out"))) \
    .withColumn("denomination",
        when(col("denomination").isNull(), 0.01)
        .when(~col("denomination").isin(valid_denoms), 0.01)
        .otherwise(col("denomination")))

records_after = df_cleansed.count()
records_removed = df_bronze.count() - records_after
print(f"  Records removed: {records_removed:,}")
print(f"  Records remaining: {records_after:,}")

# Cell 5: Deduplication
from pyspark.sql.window import Window

print("\n🔄 Removing Duplicates...")

# Remove duplicates based on event_id (keep latest ingestion)
window_spec = Window.partitionBy("event_id").orderBy(col("_bronze_ingested_at").desc())

df_deduped = df_cleansed \
    .withColumn("_row_num", row_number().over(window_spec)) \
    .filter(col("_row_num") == 1) \
    .drop("_row_num")

duplicates_removed = df_cleansed.count() - df_deduped.count()
print(f"  Duplicates removed: {duplicates_removed:,}")
print(f"  Unique records: {df_deduped.count():,}")

# Cell 6: Add Silver Metadata & Derived Columns
print("\n✨ Adding Derived Columns...")

df_silver = df_deduped \
    .withColumn("_silver_processed_at", current_timestamp()) \
    .withColumn("_silver_batch_id", lit(datetime.now().strftime("%Y%m%d_%H%M%S"))) \
    .withColumn("event_date", to_date(col("event_timestamp"))) \
    .withColumn("event_hour", hour(col("event_timestamp"))) \
    .withColumn("event_day_of_week", dayofweek(col("event_timestamp"))) \
    .withColumn("net_win", col("coin_in") - col("coin_out")) \
    .withColumn("hold_percentage",
        when(col("coin_in") > 0,
             round((col("coin_in") - col("coin_out")) / col("coin_in") * 100, 2))
        .otherwise(0)) \
    .withColumn("is_jackpot", col("event_type") == "JACKPOT") \
    .withColumn("is_game_play", col("event_type") == "GAME_PLAY")

print("  Added columns:")
print("    - _silver_processed_at (metadata)")
print("    - _silver_batch_id (metadata)")
print("    - event_date, event_hour, event_day_of_week (time dimensions)")
print("    - net_win, hold_percentage (financial KPIs)")
print("    - is_jackpot, is_game_play (flags)")

# Cell 7: Write to Silver Table
print("\n💾 Writing to Silver Table...")

df_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("event_date") \
    .saveAsTable(SILVER_TABLE)

print(f"✅ Wrote {df_silver.count():,} records to {SILVER_TABLE}")

# Cell 8: Verify Silver Table
df_verify = spark.table(SILVER_TABLE)

print("\n" + "=" * 60)
print("✅ VERIFICATION COMPLETE")
print("=" * 60)
print(f"\n📊 Silver Table Statistics:")
print(f"  Total Records: {df_verify.count():,}")
print(f"  Partitions (dates): {df_verify.select('event_date').distinct().count()}")

print(f"\n🔍 Data Quality Check:")
print(f"  Null event_ids: {df_verify.filter(col('event_id').isNull()).count()}")
print(f"  Negative coin_in: {df_verify.filter(col('coin_in') < 0).count()}")
print(f"  Future timestamps: {df_verify.filter(col('event_timestamp') > current_timestamp()).count()}")

# Display sample
print("\n📋 Sample Records:")
display(df_verify.limit(5))

# Cell 1: Configuration
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from delta.tables import DeltaTable
from datetime import datetime

BRONZE_TABLE = "lh_bronze.bronze_player_profile"
SILVER_TABLE = "silver_player_master"

# SCD Type 2 configuration
BUSINESS_KEY = "player_id"
TRACKED_COLUMNS = [
    "loyalty_tier", "points_balance", "tier_credits",
    "communication_preference", "marketing_opt_in", "account_status"
]

print("=" * 60)
print("🥈 SILVER LAYER - Player Master (SCD Type 2)")
print("=" * 60)
print(f"Business Key: {BUSINESS_KEY}")
print(f"Tracked Columns: {', '.join(TRACKED_COLUMNS)}")

# Cell 2: Read Bronze Data
df_bronze = spark.table(BRONZE_TABLE)

print(f"\n📊 Source Statistics:")
print(f"  Total Records: {df_bronze.count():,}")
print(f"  Unique Players: {df_bronze.select('player_id').distinct().count():,}")

# Cell 3: Prepare Source Data with Change Detection Hash
print("\n🔐 Creating Change Detection Hash...")

# Create hash of tracked columns for efficient change detection
df_source = df_bronze \
    .withColumn("_record_hash",
        sha2(concat_ws("|", *[coalesce(col(c).cast("string"), lit("")) for c in TRACKED_COLUMNS]), 256)) \
    .withColumn("_effective_from", current_timestamp()) \
    .withColumn("_effective_to", lit(None).cast("timestamp")) \
    .withColumn("_is_current", lit(True)) \
    .withColumn("_silver_processed_at", current_timestamp())

print("  Added SCD2 columns:")
print("    - _record_hash (change detection)")
print("    - _effective_from (version start)")
print("    - _effective_to (version end)")
print("    - _is_current (active flag)")

# Cell 4: Check if Silver table exists
table_exists = spark.catalog.tableExists(SILVER_TABLE)
print(f"\n🔍 Silver table exists: {table_exists}")

# Cell 5: Initial Load or SCD Type 2 Merge
if not table_exists:
    # Initial load - simple insert
    print("\n📥 Performing Initial Load...")

    df_source.write \
        .format("delta") \
        .mode("overwrite") \
        .saveAsTable(SILVER_TABLE)

    print(f"✅ Initial load complete: {df_source.count():,} records")

else:
    # SCD Type 2 Merge
    print("\n🔄 Performing SCD Type 2 Merge...")

    delta_table = DeltaTable.forName(spark, SILVER_TABLE)

    # Step 1: Expire changed records
    delta_table.alias("target").merge(
        df_source.alias("source"),
        f"target.{BUSINESS_KEY} = source.{BUSINESS_KEY} AND target._is_current = true"
    ).whenMatchedUpdate(
        condition="target._record_hash != source._record_hash",
        set={
            "_effective_to": "source._effective_from",
            "_is_current": "false"
        }
    ).whenNotMatchedInsertAll().execute()

    # Step 2: Insert new versions for changed records
    # Get records that were just expired
    df_expired = spark.table(SILVER_TABLE) \
        .filter(col("_is_current") == False) \
        .filter(col("_effective_to").isNotNull()) \
        .select(BUSINESS_KEY)

    # Insert new current versions
    df_new_versions = df_source.join(df_expired, BUSINESS_KEY, "inner")

    if df_new_versions.count() > 0:
        df_new_versions.write \
            .format("delta") \
            .mode("append") \
            .saveAsTable(SILVER_TABLE)
        print(f"  Inserted {df_new_versions.count():,} new versions")

    print("✅ SCD Type 2 merge complete")

# Cell 6: Verify SCD History
df_history = spark.table(SILVER_TABLE)

print("\n" + "=" * 60)
print("✅ PLAYER MASTER VERIFICATION")
print("=" * 60)
print(f"\n📊 Statistics:")
print(f"  Total Records: {df_history.count():,}")
print(f"  Current Records: {df_history.filter(col('_is_current')).count():,}")
print(f"  Historical Records: {df_history.filter(~col('_is_current')).count():,}")

# Show sample player with history
sample_player = df_history.select("player_id").first()[0]
print(f"\n📋 Sample History for Player: {sample_player}")
display(
    df_history
    .filter(col("player_id") == sample_player)
    .select("player_id", "loyalty_tier", "points_balance",
            "_effective_from", "_effective_to", "_is_current")
    .orderBy("_effective_from")
)

# Cell 1: Configuration
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from datetime import datetime

BRONZE_TABLE = "lh_bronze.bronze_financial_txn"
SILVER_TABLE = "silver_financial_reconciled"

# Regulatory thresholds
CTR_THRESHOLD = 10000  # Currency Transaction Report threshold
SAR_REVIEW_LOW = 8000  # Suspicious Activity Review lower bound
SAR_REVIEW_HIGH = 9999 # Suspicious Activity Review upper bound

print("=" * 60)
print("🥈 SILVER LAYER - Financial Transaction Reconciliation")
print("=" * 60)
print(f"CTR Threshold: ${CTR_THRESHOLD:,}")
print(f"SAR Review Range: ${SAR_REVIEW_LOW:,} - ${SAR_REVIEW_HIGH:,}")

# Cell 2: Read and Cleanse
df_bronze = spark.table(BRONZE_TABLE)

print(f"\n📊 Source Statistics:")
print(f"  Total Transactions: {df_bronze.count():,}")

df_cleansed = df_bronze \
    .filter(col("transaction_id").isNotNull()) \
    .filter(col("amount") > 0) \
    .filter(col("transaction_timestamp").isNotNull())

removed = df_bronze.count() - df_cleansed.count()
print(f"  Records Removed: {removed:,}")
print(f"  Valid Records: {df_cleansed.count():,}")

# Cell 3: Regulatory Validation
print("\n🔍 Regulatory Compliance Validation...")

df_validated = df_cleansed \
    .withColumn("ctr_threshold_check",
        when((col("amount") >= CTR_THRESHOLD) & (~col("ctr_required")), "FAIL")
        .otherwise("PASS")) \
    .withColumn("amount_validated", col("amount") > 0) \
    .withColumn("structuring_risk",
        when(col("amount").between(SAR_REVIEW_LOW, SAR_REVIEW_HIGH), "REVIEW")
        .when(col("amount") >= CTR_THRESHOLD, "CTR_REQUIRED")
        .otherwise("CLEAR"))

# Log validation results
ctr_failures = df_validated.filter(col("ctr_threshold_check") == "FAIL").count()
structuring_reviews = df_validated.filter(col("structuring_risk") == "REVIEW").count()

print(f"  CTR Validation Failures: {ctr_failures}")
print(f"  Structuring Reviews Needed: {structuring_reviews}")

# Cell 4: Add Reconciliation Columns
print("\n✨ Adding Reconciliation Metadata...")

df_silver = df_validated \
    .withColumn("_silver_processed_at", current_timestamp()) \
    .withColumn("_silver_batch_id", lit(datetime.now().strftime("%Y%m%d_%H%M%S"))) \
    .withColumn("transaction_date", to_date(col("transaction_timestamp"))) \
    .withColumn("transaction_hour", hour(col("transaction_timestamp"))) \
    .withColumn("is_cash_transaction",
        col("transaction_type").isin(["CASH_IN", "CASH_OUT", "CHIP_PURCHASE", "CHIP_REDEMPTION"])) \
    .withColumn("is_reportable", col("amount") >= CTR_THRESHOLD) \
    .withColumn("needs_review",
        (col("suspicious_activity_flag")) |
        (col("amount").between(SAR_REVIEW_LOW, SAR_REVIEW_HIGH))) \
    .withColumn("compliance_status",
        when(col("ctr_threshold_check") == "FAIL", "NON_COMPLIANT")
        .when(col("needs_review"), "PENDING_REVIEW")
        .otherwise("COMPLIANT"))

# Cell 5: Write Silver Table
print("\n💾 Writing to Silver Table...")

df_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("transaction_date") \
    .saveAsTable(SILVER_TABLE)

print(f"✅ Wrote {df_silver.count():,} records to {SILVER_TABLE}")

# Cell 6: Verification and Compliance Summary
df_verify = spark.table(SILVER_TABLE)

print("\n" + "=" * 60)
print("✅ FINANCIAL RECONCILIATION SUMMARY")
print("=" * 60)

# Compliance breakdown
print("\n📊 Compliance Status Breakdown:")
display(
    df_verify
    .groupBy("compliance_status")
    .agg(
        count("*").alias("transaction_count"),
        sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount")
    )
    .orderBy("compliance_status")
)

# Silver Layer Verification
from pyspark.sql.functions import *

tables = [
    "silver_slot_cleansed",
    "silver_player_master",
    "silver_financial_reconciled",
    "silver_table_enriched",
    "silver_security_enriched",
    "silver_compliance_validated"
]

print("=" * 70)
print("🥈 SILVER LAYER VERIFICATION REPORT")
print("=" * 70)
print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("-" * 70)

total_records = 0
passed = 0
failed = 0

for table in tables:
    try:
        df = spark.table(table)
        count = df.count()
        total_records += count

        # Verify required metadata columns
        has_processed = "_silver_processed_at" in df.columns
        has_batch = "_silver_batch_id" in df.columns or "_record_hash" in df.columns

        if has_processed:
            status = "✅ PASS"
            passed += 1
        else:
            status = "⚠️ WARN"

        print(f"{status} {table:40} {count:>12,} records")

    except Exception as e:
        print(f"❌ FAIL {table:40} NOT FOUND")
        failed += 1

print("-" * 70)
print(f"{'SUMMARY':40} {total_records:>12,} total records")
print(f"{'Tables Passed:':40} {passed}/{len(tables)}")
if failed > 0:
    print(f"{'Tables Failed:':40} {failed}/{len(tables)}")
print("=" * 70)

# quality_rules.py - Data Quality Rule Definitions
QUALITY_RULES = {
    "slot_telemetry": {
        "completeness": {
            "required_fields": ["event_id", "machine_id", "event_type", "event_timestamp"],
            "threshold": 0.99  # 99% completeness required
        },
        "validity": {
            "coin_in": {"rule": "value >= 0", "description": "Coin-in must be non-negative"},
            "coin_out": {"rule": "value >= 0", "description": "Coin-out must be non-negative"},
            "denomination": {
                "rule": "value in [0.01, 0.05, 0.25, 0.50, 1.00, 5.00, 25.00, 100.00]",
                "description": "Valid denomination values"
            }
        },
        "uniqueness": {
            "primary_key": ["event_id"],
            "threshold": 1.0  # 100% unique
        },
        "timeliness": {
            "rule": "event_timestamp <= current_timestamp",
            "max_age_days": 7
        }
    },
    "player_profile": {
        "completeness": {
            "required_fields": ["player_id", "loyalty_number"],
            "threshold": 0.99
        },
        "validity": {
            "loyalty_tier": {
                "rule": "value in ['Bronze', 'Silver', 'Gold', 'Platinum', 'Diamond']",
                "description": "Valid loyalty tiers"
            },
            "points_balance": {"rule": "value >= 0", "description": "Non-negative points"}
        },
        "uniqueness": {
            "primary_key": ["player_id"],
            "threshold": 1.0
        }
    },
    "financial_txn": {
        "completeness": {
            "required_fields": ["transaction_id", "amount", "transaction_type"],
            "threshold": 0.999  # 99.9% for financial data
        },
        "validity": {
            "amount": {"rule": "value > 0", "description": "Positive transaction amount"},
            "ctr_compliance": {
                "rule": "if amount >= 10000 then ctr_required = true",
                "description": "CTR threshold compliance"
            }
        },
        "uniqueness": {
            "primary_key": ["transaction_id"],
            "threshold": 1.0
        }
    }
}

def calculate_quality_score(df, rules):
    """
    Calculate overall data quality score based on multiple dimensions.

    Returns a score from 0-100 where:
    - 95-100: Excellent quality
    - 85-94:  Good quality
    - 70-84:  Acceptable quality
    - <70:    Needs attention
    """
    total_records = df.count()
    if total_records == 0:
        return 0.0

    scores = []

    # Completeness score
    if "completeness" in rules:
        null_counts = [df.filter(col(f).isNull()).count()
                       for f in rules["completeness"]["required_fields"]]
        completeness = 1 - (sum(null_counts) / (total_records * len(null_counts)))
        scores.append(completeness * 100)

    # Uniqueness score
    if "uniqueness" in rules:
        unique_count = df.select(rules["uniqueness"]["primary_key"]).distinct().count()
        uniqueness = unique_count / total_records
        scores.append(uniqueness * 100)

    # Return weighted average
    return sum(scores) / len(scores) if scores else 0.0

# Get watermark from last successful run
last_processed = spark.sql(f"""
    SELECT MAX(_bronze_ingested_at) as watermark
    FROM {SILVER_TABLE}
""").first()["watermark"]

# Process only new records (incremental)
df_incremental = df_bronze.filter(
    col("_bronze_ingested_at") > last_processed
)

print(f"Processing {df_incremental.count():,} new records since {last_processed}")

# Log quality metrics to monitoring table
from datetime import datetime

quality_metrics = {
    "table_name": SILVER_TABLE,
    "processed_at": datetime.now().isoformat(),
    "records_processed": df_silver.count(),
    "records_rejected": failures,
    "quality_score": round((df_silver.count() - failures) / df_silver.count() * 100, 2),
    "batch_id": BATCH_ID
}

# Write to monitoring table
spark.createDataFrame([quality_metrics]).write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("data_quality_log")

print(f"📊 Quality Score: {quality_metrics['quality_score']}%")

# Separate good records from bad
df_good = df_bronze.filter(quality_conditions)
df_bad = df_bronze.filter(~quality_conditions)

# Write bad records to quarantine for investigation
if df_bad.count() > 0:
    df_bad \
        .withColumn("_quarantine_reason", lit("Quality check failed")) \
        .withColumn("_quarantine_timestamp", current_timestamp()) \
        .write \
        .format("delta") \
        .mode("append") \
        .saveAsTable("quarantine_slot_telemetry")

    print(f"⚠️ Quarantined {df_bad.count():,} records for investigation")

# Compare schemas to identify differences
bronze_cols = set(df_bronze.columns)
silver_cols = set(spark.table(SILVER_TABLE).columns)

new_columns = bronze_cols - silver_cols
removed_columns = silver_cols - bronze_cols

print(f"New columns in Bronze: {new_columns}")
print(f"Columns only in Silver: {removed_columns}")

# Enable schema evolution if needed
df.write \
    .option("mergeSchema", "true") \
    .saveAsTable(SILVER_TABLE)

# Debug SCD2: Check for hash changes
df_current = spark.table(SILVER_TABLE).filter(col("_is_current"))
df_new = df_source

df_comparison = df_current.alias("curr").join(
    df_new.alias("new"),
    col("curr.player_id") == col("new.player_id"),
    "inner"
).select(
    col("curr.player_id"),
    col("curr._record_hash").alias("current_hash"),
    col("new._record_hash").alias("new_hash"),
    (col("curr._record_hash") != col("new._record_hash")).alias("has_changed")
)

print(f"Records with changes: {df_comparison.filter(col('has_changed')).count()}")