Home > Notebooks > Gold Layer¶

# Star Schema - Fact Table Pattern
from pyspark.sql.functions import sum, avg, count, date_trunc

fact_slot_performance = (spark.table("silver.slot_cleansed")
    # Aggregate to grain (machine, hour)
    .groupBy(
        col("machine_id"),
        date_trunc("hour", col("spin_timestamp")).alias("hour_key"),
        col("denomination")
    )
    .agg(
        count("*").alias("total_spins"),
        sum("wager_amount").alias("total_coin_in"),
        sum("payout_amount").alias("total_coin_out"),
        (sum("wager_amount") - sum("payout_amount")).alias("net_win"),
        avg("session_duration_seconds").alias("avg_session_duration")
    )
    # Calculate KPIs
    .withColumn("hold_percentage", 
        (col("net_win") / col("total_coin_in") * 100).cast("decimal(5,2)"))
    .withColumn("theoretical_rtp",
        (col("total_coin_out") / col("total_coin_in") * 100).cast("decimal(5,2)"))
)

# Write with V-Order optimization for Direct Lake
fact_slot_performance.write \
    .format("delta") \
    .option("delta.parquet.vorder.enabled", "true") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_slot_performance")

# SCD Type 1 Dimension with Business Attributes
dim_player_360 = (spark.table("silver.player_master")
    .select(
        col("player_id").alias("player_key"),
        col("player_id"),
        col("first_name"),
        col("last_name"),
        col("player_tier"),
        col("enrollment_date"),
        col("preferred_denomination"),
        col("home_casino")
    )
    # Add calculated attributes
    .join(lifetime_value_df, "player_id", "left")
    .join(visit_frequency_df, "player_id", "left")
    .withColumn("player_segment", 
        when(col("lifetime_value") > 100000, "VIP")
        .when(col("lifetime_value") > 10000, "HIGH")
        .when(col("lifetime_value") > 1000, "MEDIUM")
        .otherwise("CASUAL"))
)

# Validate KPI calculations
kpi_validation = spark.sql("""
    SELECT 
        SUM(CASE WHEN hold_percentage < 0 OR hold_percentage > 20 THEN 1 ELSE 0 END) as invalid_hold,
        SUM(CASE WHEN theoretical_rtp < 80 OR theoretical_rtp > 100 THEN 1 ELSE 0 END) as invalid_rtp
    FROM gold.fact_slot_performance
""").first()
assert kpi_validation.invalid_hold == 0, "Invalid hold percentages found!"
assert kpi_validation.invalid_rtp == 0, "Invalid RTP values found!"

# Verify totals match Silver layer
gold_total = spark.sql("SELECT SUM(total_coin_in) FROM gold.fact_slot_performance").first()[0]
silver_total = spark.sql("SELECT SUM(wager_amount) FROM silver.slot_cleansed").first()[0]
variance = abs(gold_total - silver_total) / silver_total
assert variance < 0.0001, f"Variance {variance:.4%} exceeds threshold!"

# Check Direct Lake compatibility
from delta import DeltaTable
dt = DeltaTable.forName(spark, "gold.fact_slot_performance")
history = dt.history(1).select("operationMetrics").first()
print(f"Files: {history.operationMetrics.get('numFiles', 'N/A')}")

# Enable V-Order for optimal Direct Lake performance
spark.conf.set("spark.databricks.delta.properties.defaults.parquet.vorder.enabled", "true")

# Optimize file sizes (target 128MB-256MB)
spark.sql("OPTIMIZE gold.fact_slot_performance ZORDER BY (date_key, machine_key)")

# Vacuum old versions (maintain 7 days for time travel)
spark.sql("VACUUM gold.fact_slot_performance RETAIN 168 HOURS")