🔥 Spark Job Definitions - Production-Grade Spark in Fabric¶

"""
bronze_slot_telemetry_sjd.py
Production SJD for ingesting slot telemetry data into the bronze layer.
"""
import argparse
import logging
import sys
from datetime import datetime, timedelta

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("bronze_slot_telemetry")


def parse_args(args=None):
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(description="Bronze slot telemetry ingestion")
    parser.add_argument("--date", required=True, help="Processing date (YYYY-MM-DD)")
    parser.add_argument("--env", default="dev", choices=["dev", "staging", "prod"])
    parser.add_argument("--source-path", required=True, help="Source data path in OneLake")
    parser.add_argument("--target-table", default="slot_telemetry", help="Target Delta table")
    parser.add_argument("--max-records", type=int, default=0, help="Limit records (0=unlimited)")
    parser.add_argument("--dry-run", action="store_true", help="Validate without writing")
    return parser.parse_args(args)


def define_schema():
    """Define the expected source schema."""
    return StructType([
        StructField("machine_id", StringType(), False),
        StructField("casino_id", StringType(), False),
        StructField("event_timestamp", TimestampType(), False),
        StructField("event_type", StringType(), False),
        StructField("bet_amount", DoubleType(), True),
        StructField("win_amount", DoubleType(), True),
        StructField("player_id", StringType(), True),
        StructField("denomination", DoubleType(), True),
        StructField("zone_id", StringType(), True),
    ])


def validate_data(df, date_str):
    """Run quality checks on ingested data."""
    total = df.count()
    nulls = df.filter(F.col("machine_id").isNull() | F.col("event_timestamp").isNull()).count()

    if total == 0:
        raise ValueError(f"No records found for date {date_str}")

    null_rate = nulls / total
    if null_rate > 0.05:
        raise ValueError(f"Null rate {null_rate:.1%} exceeds 5% threshold")

    logger.info(f"Validation passed: {total} records, {null_rate:.1%} null rate")
    return total


def ingest(spark, args):
    """Main ingestion logic."""
    logger.info(f"Starting bronze ingestion for {args.date} in {args.env}")

    schema = define_schema()

    # Read source data with schema enforcement
    df = (
        spark.read
        .schema(schema)
        .format("parquet")
        .load(f"{args.source_path}/date={args.date}")
    )

    if args.max_records > 0:
        df = df.limit(args.max_records)

    # Add metadata columns
    df = df.withColumns({
        "_ingestion_timestamp": F.current_timestamp(),
        "_source_file": F.input_file_name(),
        "_processing_date": F.lit(args.date),
        "_environment": F.lit(args.env),
    })

    # Validate
    record_count = validate_data(df, args.date)

    if args.dry_run:
        logger.info(f"DRY RUN: Would write {record_count} records to {args.target_table}")
        df.show(5)
        return

    # Write to Delta table (bronze: append-only)
    target_path = f"abfss://bronze@onelake.dfs.fabric.microsoft.com/lh_bronze.Lakehouse/Tables/{args.target_table}"

    (
        df.write
        .format("delta")
        .mode("append")
        .partitionBy("_processing_date")
        .save(target_path)
    )

    logger.info(f"Successfully wrote {record_count} records to {args.target_table}")


def main():
    args = parse_args()

    spark = (
        SparkSession.builder
        .appName(f"bronze-slot-telemetry-{args.env}-{args.date}")
        .getOrCreate()
    )

    try:
        ingest(spark, args)
    except Exception as e:
        logger.error(f"Ingestion failed: {e}", exc_info=True)
        sys.exit(1)
    finally:
        spark.stop()


if __name__ == "__main__":
    main()

# In main.py, reference files are available on the Spark driver
import json
import os

# Reference files are placed in the working directory
with open("config.json") as f:
    config = json.load(f)

# Import reference Python modules
from utils import setup_logging
from schemas import SLOT_SCHEMA
from validators import validate_bronze

import argparse

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--date", required=True)
    parser.add_argument("--env", default="dev")
    parser.add_argument("--table", required=True)
    parser.add_argument("--parallelism", type=int, default=4)
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()

# Read Spark conf values set at SJD or environment level
env = spark.conf.get("spark.fabric.env", "dev")
workspace = spark.conf.get("spark.fabric.workspace.id")

import sys
import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    reraise=True
)
def read_source_data(spark, path, schema):
    """Read source data with retry for transient failures."""
    try:
        return spark.read.schema(schema).parquet(path)
    except Exception as e:
        logger.warning(f"Read attempt failed: {e}")
        raise

def main():
    args = parse_args()
    spark = SparkSession.builder.getOrCreate()

    try:
        df = read_source_data(spark, args.source_path, SCHEMA)
        process(df, args)
    except FileNotFoundError:
        logger.error(f"Source path not found: {args.source_path}")
        sys.exit(2)  # Distinct exit code for missing data
    except ValueError as e:
        logger.error(f"Data validation failed: {e}")
        sys.exit(3)  # Distinct exit code for quality failure
    except Exception as e:
        logger.error(f"Unexpected error: {e}", exc_info=True)
        sys.exit(1)
    finally:
        spark.stop()

# 1. Partition reads to avoid full scans
df = spark.read.format("delta").load(path).filter(F.col("date") == target_date)

# 2. Use broadcast for small lookup tables
zones = spark.read.format("delta").load(zone_path)
df = df.join(F.broadcast(zones), "zone_id", "left")

# 3. Repartition before wide transformations
df = df.repartition(args.parallelism, "casino_id")

# 4. Cache intermediate results used multiple times
validated = df.filter(F.col("_valid") == True).cache()
write_silver(validated)
write_quality_report(validated)
validated.unpersist()

# 5. Use adaptive query execution (enabled by default in Fabric)
spark.conf.set("spark.sql.adaptive.enabled", "true")

# fabric-cicd-deploy.py (SJD deployment)
from fabric_cicd import FabricClient

client = FabricClient(workspace_id="your-workspace-id")

# Deploy SJD from local files
client.deploy_item(
    item_type="SparkJobDefinition",
    display_name="bronze-slot-telemetry-sjd",
    definition_path="spark-jobs/bronze_slot_telemetry/",
    environment_name="casino-bronze-env"
)

# tests/test_bronze_slot_telemetry.py
import pytest
from pyspark.sql import SparkSession
from bronze_slot_telemetry_sjd import parse_args, define_schema, validate_data, ingest

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.master("local[2]").getOrCreate()

def test_parse_args():
    args = parse_args(["--date", "2026-01-01", "--env", "dev", "--source-path", "/data"])
    assert args.date == "2026-01-01"
    assert args.env == "dev"
    assert args.dry_run is False

def test_schema():
    schema = define_schema()
    field_names = [f.name for f in schema.fields]
    assert "machine_id" in field_names
    assert "event_timestamp" in field_names

def test_validate_data_empty(spark):
    df = spark.createDataFrame([], define_schema())
    with pytest.raises(ValueError, match="No records found"):
        validate_data(df, "2026-01-01")

def test_validate_data_high_nulls(spark):
    from pyspark.sql import Row
    # Create data with >5% null machine_id
    rows = [Row(machine_id=None, casino_id="C1", event_timestamp=None,
                event_type="spin", bet_amount=1.0, win_amount=0.0,
                player_id="P1", denomination=0.25, zone_id="Z1")]
    df = spark.createDataFrame(rows, define_schema())
    with pytest.raises(ValueError, match="Null rate"):
        validate_data(df, "2026-01-01")