Local PySpark Debugging for Fabric Development¶

# verify_spark.py
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("verify")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
    .getOrCreate()
)

df = spark.range(10)
df.show()
print(f"Spark version: {spark.version}")
print(f"Python: {spark.sparkContext.pythonVer}")
spark.stop()
print("Local Spark is working.")

import os

try:
    from notebookutils import mssparkutils  # noqa: F401
    IS_FABRIC = True
except ImportError:
    IS_FABRIC = False


def get_secret(vault_url: str, secret_name: str) -> str:
    """Retrieve a secret from Key Vault (Fabric) or env var (local)."""
    if IS_FABRIC:
        return mssparkutils.credentials.getSecret(vault_url, secret_name)
    value = os.environ.get(secret_name.upper().replace("-", "_"))
    if value is None:
        raise ValueError(
            f"Secret '{secret_name}' not found. "
            f"Set env var {secret_name.upper().replace('-', '_')} for local dev."
        )
    return value


def get_source_path(fabric_path: str, local_path: str) -> str:
    """Return the appropriate path based on execution environment."""
    if IS_FABRIC:
        return fabric_path
    return local_path

# validation/conftest.py
import sys
from unittest.mock import MagicMock

# Register mssparkutils mock before any test imports notebook code
mock_mssparkutils = MagicMock()
mock_mssparkutils.credentials.getSecret.return_value = "mock-secret"
mock_mssparkutils.env.getWorkspaceName.return_value = "test-workspace"
mock_mssparkutils.env.getLakehouseName.return_value = "lh_bronze"

sys.modules["notebookutils"] = MagicMock()
sys.modules["notebookutils.mssparkutils"] = mock_mssparkutils

"""
scripts/create_local_fixtures.py
=================================
Generate small sample datasets for local Spark debugging.
"""

import os
import sys
from pathlib import Path

# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent))

from data_generation.generators.slot_telemetry_generator import SlotTelemetryGenerator
from data_generation.generators.player_generator import PlayerGenerator


def create_fixtures(output_dir: str = "temp/fixtures") -> None:
    """Generate small sample datasets for local debugging."""
    os.makedirs(output_dir, exist_ok=True)

    # Slot telemetry: 100 records
    slot_gen = SlotTelemetryGenerator()
    slot_data = slot_gen.generate_batch(100)
    slot_gen.save_parquet(slot_data, f"{output_dir}/slot_telemetry.parquet")
    print(f"Created {output_dir}/slot_telemetry.parquet (100 records)")

    # Players: 50 records
    player_gen = PlayerGenerator()
    player_data = player_gen.generate_batch(50)
    player_gen.save_parquet(player_data, f"{output_dir}/players.parquet")
    print(f"Created {output_dir}/players.parquet (50 records)")


if __name__ == "__main__":
    create_fixtures()

from datetime import datetime
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    DoubleType, StringType, StructField, StructType, TimestampType,
)

spark = SparkSession.builder.master("local[2]").getOrCreate()

# Minimal test DataFrame
schema = StructType([
    StructField("machine_id", StringType(), False),
    StructField("amount", DoubleType(), True),
    StructField("timestamp", TimestampType(), False),
])

data = [
    ("SLOT-001", 25.0, datetime(2026, 1, 1, 10, 0)),
    ("SLOT-002", 12500.0, datetime(2026, 1, 1, 10, 5)),
    ("SLOT-003", 9500.0, datetime(2026, 1, 1, 10, 10)),
]

df = spark.createDataFrame(data, schema=schema)
df.show()

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
    .getOrCreate()
)

# Write Delta table locally
df.write.format("delta").mode("overwrite").save("temp/delta/slot_telemetry")

# Read it back
df_read = spark.read.format("delta").load("temp/delta/slot_telemetry")
df_read.show()

# Time travel
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("temp/delta/slot_telemetry")

df.show(5, truncate=False)
df.printSchema()
df.count()
df.describe().show()

df.count() > 100

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
    .getOrCreate()
)

# ── Write ──────────────────────────────────────────────────────
df.write.format("delta").mode("overwrite").save("temp/delta/bronze_slots")

# ── Read ───────────────────────────────────────────────────────
df = spark.read.format("delta").load("temp/delta/bronze_slots")

# ── Append ─────────────────────────────────────────────────────
new_data.write.format("delta").mode("append").save("temp/delta/bronze_slots")

# ── Merge (upsert) ────────────────────────────────────────────
from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "temp/delta/silver_players")
target.alias("t").merge(
    source=updates.alias("s"),
    condition="t.player_id = s.player_id",
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

# ── Time travel ────────────────────────────────────────────────
df_v0 = (
    spark.read.format("delta")
    .option("versionAsOf", 0)
    .load("temp/delta/bronze_slots")
)

# ── History ────────────────────────────────────────────────────
delta_table = DeltaTable.forPath(spark, "temp/delta/bronze_slots")
delta_table.history().show(truncate=False)

# ── Schema evolution ──────────────────────────────────────────
df_new_cols.write.format("delta").mode("append") \
    .option("mergeSchema", "true") \
    .save("temp/delta/bronze_slots")

# Bronze: raw ingestion
raw_df = spark.read.parquet("temp/fixtures/slot_telemetry.parquet")
raw_df.write.format("delta").mode("overwrite").save("temp/delta/bronze/slot_telemetry")

# Silver: cleansed
bronze_df = spark.read.format("delta").load("temp/delta/bronze/slot_telemetry")
silver_df = (
    bronze_df
    .filter(col("machine_id").isNotNull())
    .dropDuplicates(["machine_id", "timestamp"])
    .withColumn("ingested_at", current_timestamp())
)
silver_df.write.format("delta").mode("overwrite").save("temp/delta/silver/slot_telemetry")

# Gold: aggregated
silver_df = spark.read.format("delta").load("temp/delta/silver/slot_telemetry")
gold_df = (
    silver_df
    .groupBy("machine_id")
    .agg(
        sum("amount").alias("total_amount"),
        count("*").alias("event_count"),
        avg("amount").alias("avg_amount"),
    )
)
gold_df.write.format("delta").mode("overwrite").save("temp/delta/gold/slot_performance")

"""notebooks/utils/path_resolver.py"""

import os

# Detect execution environment
IS_FABRIC = "FABRIC_RUNTIME_VERSION" in os.environ


def resolve_path(fabric_path: str, local_path: str) -> str:
    """Return the appropriate path for the current environment."""
    return fabric_path if IS_FABRIC else local_path


# Usage in notebooks:
source = resolve_path(
    fabric_path="Files/output/bronze_slot_telemetry.parquet",
    local_path="temp/fixtures/slot_telemetry.parquet",
)

import duckdb

# Read Parquet directly
result = duckdb.sql("""
    SELECT machine_id, COUNT(*) as event_count, SUM(amount) as total
    FROM 'temp/fixtures/slot_telemetry.parquet'
    GROUP BY machine_id
    ORDER BY total DESC
    LIMIT 10
""")
result.show()

# Read Delta tables (DuckDB 0.10+)
result = duckdb.sql("""
    SELECT * FROM delta_scan('temp/delta/bronze/slot_telemetry')
    WHERE amount >= 10000
""")
result.show()

# Validate SQL logic before porting to Spark
ctr_check = duckdb.sql("""
    SELECT
        machine_id,
        amount,
        CASE WHEN amount >= 10000 THEN TRUE ELSE FALSE END AS ctr_flag
    FROM 'temp/fixtures/slot_telemetry.parquet'
    WHERE amount >= 8000
    ORDER BY amount DESC
""")
ctr_check.show()

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .config("spark.hadoop.fs.s3a.endpoint", "http://localhost:9000")
    .config("spark.hadoop.fs.s3a.access.key", "fabricadmin")
    .config("spark.hadoop.fs.s3a.secret.key", "fabricpassword123")
    .config("spark.hadoop.fs.s3a.path.style.access", "true")
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
    .getOrCreate()
)

# Write to MinIO (simulating OneLake write)
df.write.format("delta").mode("overwrite").save("s3a://lakehouse/bronze/slot_telemetry")

# Read from MinIO
df = spark.read.format("delta").load("s3a://lakehouse/bronze/slot_telemetry")

df = spark.read.format("delta").load("temp/delta/bronze/slot_telemetry")
# <-- breakpoint here

# Debug Console
df.printSchema()
# Output shows column is named 'timestamp', not 'event_timestamp'

test_data = [
    ("M1", 9999.99),   # Below threshold
    ("M2", 10000.00),   # At threshold (should flag)
    ("M3", 10000.01),   # Above threshold (should flag)
]

spark = (
    SparkSession.builder
    .master("local[2]")            # Use 2 cores, not local[*]
    .config("spark.ui.enabled", "false")          # Skip Spark UI
    .config("spark.driver.memory", "2g")          # Limit memory
    .config("spark.sql.shuffle.partitions", "2")  # Fewer partitions
    .config("spark.default.parallelism", "2")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)