Cheat Sheets¶

# ---- Delta Lake (primary format) ----

# Read Delta table by name (default Lakehouse)
df = spark.table("bronze_slot_telemetry")

# Read Delta table by path
df = spark.read.format("delta").load("Tables/bronze_slot_telemetry")

# Write Delta table (overwrite)
df.write.format("delta").mode("overwrite").saveAsTable("silver_slot_cleansed")

# Write Delta table (append)
df.write.format("delta").mode("append").saveAsTable("bronze_slot_telemetry")

# Write with schema evolution
df.write.format("delta").mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("bronze_slot_telemetry")

# Write with V-Order (required for Direct Lake performance)
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")
df.write.format("delta").mode("overwrite").saveAsTable("gold_slot_performance")

# ---- Parquet (landing zone) ----

# Read Parquet from Files section
df = spark.read.parquet("Files/landing/slot_telemetry/*.parquet")

# Read with explicit schema
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("amount", DoubleType(), True)
])
df = spark.read.schema(schema).parquet("Files/landing/data.parquet")

# ---- CSV ----
df = spark.read.option("header", "true").option("inferSchema", "true").csv("Files/landing/data.csv")

# ---- JSON ----
df = spark.read.option("multiline", "true").json("Files/landing/data.json")

# Check schema
df.printSchema()
df.dtypes  # List of (column_name, type_string) tuples

# Cast columns
from pyspark.sql.functions import col
df = df.withColumn("amount", col("amount").cast("double"))

# Drop columns
df = df.drop("unwanted_column")

# Rename columns
df = df.withColumnRenamed("old_name", "new_name")

# Add NOT NULL constraint (Delta)
spark.sql("ALTER TABLE silver_financial ADD CONSTRAINT amount_nn CHECK (amount IS NOT NULL)")

# Add CHECK constraint
spark.sql("ALTER TABLE silver_financial ADD CONSTRAINT amount_pos CHECK (amount >= 0)")

from pyspark.sql.functions import (
    col, lit, when, coalesce, concat, upper, lower, trim,
    year, month, dayofmonth, hour, date_format, current_timestamp,
    sum, avg, count, min, max, countDistinct,
    row_number, rank, dense_rank, lag, lead,
    explode, arrays_zip, from_json, to_json,
    sha2, md5, regexp_replace,
    broadcast
)
from pyspark.sql.window import Window

# ---- Filtering ----
df = df.filter(col("amount") > 10000)                   # CTR threshold
df = df.filter(col("event_date").between("2025-01-01", "2025-12-31"))

# ---- Window Functions ----
window_spec = Window.partitionBy("player_id").orderBy(col("event_ts").desc())
df = df.withColumn("row_num", row_number().over(window_spec))           # Dedup
df = df.withColumn("prev_amount", lag("amount", 1).over(window_spec))   # Previous value
df = df.withColumn("running_total", sum("amount").over(
    Window.partitionBy("player_id").orderBy("event_ts").rowsBetween(Window.unboundedPreceding, 0)
))

# ---- Pivot ----
pivot_df = df.groupBy("property_id").pivot("game_type").agg(sum("amount"))

# ---- Explode (flatten arrays) ----
df = df.withColumn("tag", explode(col("tags")))

# ---- UDF (use sparingly -- prefer built-in functions) ----
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def mask_ssn(ssn):
    return f"XXX-XX-{ssn[-4:]}" if ssn and len(ssn) >= 4 else None

df = df.withColumn("masked_ssn", mask_ssn(col("ssn")))

# ---- MERGE / Upsert (Delta) ----
from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "silver_player_master")
target.alias("t").merge(
    source_df.alias("s"),
    "t.player_id = s.player_id"
).whenMatchedUpdate(set={
    "name": "s.name",
    "tier": "s.tier",
    "updated_at": "s.updated_at"
}).whenNotMatchedInsertAll().execute()

# ---- Optimize / Vacuum ----
spark.sql("OPTIMIZE gold_slot_performance ZORDER BY (property_id, event_date)")
spark.sql("OPTIMIZE gold_slot_performance USING VORDER")
spark.sql("VACUUM gold_slot_performance RETAIN 168 HOURS")

# ---- File System ----
mssparkutils.fs.ls("Files/landing/")                     # List files
mssparkutils.fs.mkdirs("Files/archive/2025/")            # Create directory
mssparkutils.fs.cp("Files/a.parquet", "Files/b.parquet")  # Copy file
mssparkutils.fs.mv("Files/a.parquet", "Files/archive/")   # Move file
mssparkutils.fs.rm("Files/temp/", True)                   # Delete recursively
mssparkutils.fs.head("Files/config.json", 1000)           # Read first 1000 bytes

# ---- Notebook Orchestration ----
mssparkutils.notebook.run("01_bronze_slot_telemetry", timeout_seconds=600,
                          arguments={"batch_date": "2025-01-15"})

# Run multiple notebooks in parallel
mssparkutils.notebook.runMultiple([
    {"notebook": "01_bronze_slot_telemetry", "args": {"batch_date": "2025-01-15"}},
    {"notebook": "02_bronze_player_profile", "args": {"batch_date": "2025-01-15"}}
])

# Exit notebook with value (for pipeline return)
mssparkutils.notebook.exit("SUCCESS")

# ---- Credentials ----
token = mssparkutils.credentials.getToken("storage")      # Get access token
secret = mssparkutils.credentials.getSecret("kv-name", "secret-name")  # Key Vault

# ---- Lakehouse ----
mssparkutils.lakehouse.list()                              # List Lakehouses
mssparkutils.lakehouse.get("lh_bronze")                    # Get Lakehouse details

# ---- Performance Tuning ----
spark.conf.set("spark.sql.shuffle.partitions", "8")             # POC scale (default 200)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10485760")  # 10 MB
spark.conf.set("spark.sql.adaptive.enabled", "true")            # AQE (default in Fabric)
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# ---- V-Order for Direct Lake ----
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

# ---- Delta Lake ----
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# ---- Memory ----
spark.conf.set("spark.driver.maxResultSize", "2g")              # Limit driver result size