🎰 Casino Analytics Cheat Sheet¶

# Filter high-value transactions
df.filter(col("amount") >= 10000)

# Select & rename columns
df.select(col("player_id"), col("amount").alias("wager_amount"))

# Group and aggregate
df.groupBy("casino_id", "game_type").agg(
    sum("amount").alias("total_wagered"),
    count("*").alias("transaction_count"),
    avg("amount").alias("avg_wager")
)

# Window functions (player ranking)
from pyspark.sql.window import Window
window = Window.partitionBy("casino_id").orderBy(desc("total_wagered"))
df.withColumn("rank", rank().over(window))

# Date extraction
df.withColumn("play_date", to_date(col("timestamp")))
df.withColumn("play_hour", hour(col("timestamp")))

# Join tables
players.join(transactions, "player_id", "left")

# Count nulls per column
df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns])

# Distinct counts
df.select(countDistinct("player_id").alias("unique_players"))

# Check for duplicates
df.groupBy("transaction_id").count().filter(col("count") > 1)

from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "Tables/silver_player_profile")
source = spark.read.format("delta").load("Tables/bronze_player")

target.alias("t").merge(
    source.alias("s"),
    "t.player_id = s.player_id"
).whenMatchedUpdate(set={
    "email": "s.email",
    "tier_status": "s.tier_status",
    "updated_at": "current_timestamp()"
}).whenNotMatchedInsertAll().execute()

# Add audit columns for slowly changing dimensions
df.withColumn("effective_from", current_timestamp()) \
  .withColumn("effective_to", lit(None).cast("timestamp")) \
  .withColumn("is_current", lit(True))

# Append-only, minimal transformation
df_raw = spark.read.json("Files/raw/slot_events/")

df_bronze = df_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_file", input_file_name())

df_bronze.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", True) \
    .save("Tables/bronze_slot_telemetry")

# Schema enforcement, deduplication, null handling
df_silver = spark.read.format("delta").load("Tables/bronze_slot_telemetry") \
    .filter(col("player_id").isNotNull()) \
    .dropDuplicates(["transaction_id"]) \
    .withColumn("amount", col("amount").cast("decimal(18,2)")) \
    .withColumn("player_id", upper(trim(col("player_id"))))

df_silver.write.format("delta") \
    .mode("overwrite") \
    .save("Tables/silver_slot_activity")

# Business-level aggregations
df_gold = spark.read.format("delta").load("Tables/silver_slot_activity") \
    .groupBy("casino_id", "game_type", to_date("timestamp").alias("play_date")) \
    .agg(
        sum("amount").alias("total_wagered"),
        sum("win_amount").alias("total_payouts"),
        countDistinct("player_id").alias("unique_players"),
        count("*").alias("total_spins")
    ) \
    .withColumn("hold_percentage", 
        (col("total_wagered") - col("total_payouts")) / col("total_wagered") * 100)

df_gold.write.format("delta") \
    .mode("overwrite") \
    .save("Tables/gold_daily_gaming_summary")

# CTR candidates
df.filter(col("amount") >= 10000) \
  .filter(col("transaction_type").isin("CASH_IN", "CHIP_PURCHASE"))

# Structuring detection (24-hour rolling)
from pyspark.sql.window import Window
window_24h = Window.partitionBy("player_id").orderBy("timestamp").rangeBetween(-86400, 0)
df.withColumn("rolling_24h_total", sum("amount").over(window_24h)) \
  .filter((col("rolling_24h_total") >= 8000) & (col("rolling_24h_total") < 10000))

# W-2G qualifying wins
df.filter(
    ((col("game_type") == "SLOT") & (col("win_amount") >= 1200)) |
    ((col("game_type") == "KENO") & (col("win_amount") >= 1500)) |
    ((col("game_type") == "POKER") & (col("win_amount") >= 5000))
)

# Check available tables
spark.catalog.listTables("bronze")

# View table schema
spark.read.format("delta").load("Tables/bronze_slot_telemetry").printSchema()

# Check file counts
mssparkutils.fs.ls("Tables/bronze_slot_telemetry")

# Clear cache
spark.catalog.clearCache()

# Check executor memory
spark.sparkContext.getConf().get("spark.executor.memory")

# Check table history
spark.sql("DESCRIBE HISTORY bronze_slot_telemetry LIMIT 10").show()

# Check table details
spark.sql("DESCRIBE DETAIL bronze_slot_telemetry").show()

# Count files (detect small file problem)
spark.sql("DESCRIBE DETAIL bronze_slot_telemetry") \
    .select("numFiles", "sizeInBytes").show()

# Repair corrupt table
spark.sql("FSCK REPAIR TABLE bronze_slot_telemetry")

# Optimize table (compact small files)
spark.sql("OPTIMIZE bronze_slot_telemetry")

# Add Z-ORDER for common filters
spark.sql("OPTIMIZE silver_player_activity ZORDER BY (player_id, play_date)")

# Analyze table statistics
spark.sql("ANALYZE TABLE gold_daily_gaming_summary COMPUTE STATISTICS")

# Check partition layout
spark.read.format("delta").load("Tables/silver_transactions").inputFiles()