Spark Operations Cli

Update Check — ONCE PER SESSION (mandatory)

The first time this skill is used in a session, run the check-updates skill before proceeding.

- GitHub Copilot CLI / VS Code: invoke the check-updates skill.

- Claude Code / Cowork / Cursor / Windsurf / Codex: compare local vs remote package.json version.

- Skip if the check was already performed earlier in this session.

CRITICAL NOTES

1. To find the workspace details (including its ID) from workspace name: list all workspaces and, then, use JMESPath filtering

2. To find the item details (including its ID) from workspace ID, item type, and item name: list all items of that type in that workspace and, then, use JMESPath filtering

3. Skill disambiguation: spark-operations-cli is for read-only triage and diagnosis of existing jobs and sessions. For creating notebooks, running new jobs, or Spark development, use spark-authoring-cli. For interactive PySpark analysis and Livy session creation, use spark-consumption-cli.

Spark Operations — CLI Skill

This skill provides diagnostics for Microsoft Fabric Spark job failures, Livy session health, and performance bottlenecks using Fabric REST APIs and CLI tools (az rest). All diagnostic operations are read-only; session cleanup (e.g., stopping zombie sessions) requires explicit user confirmation. For Spark development and notebook authoring, use spark-authoring-cli. For interactive PySpark analysis, use spark-consumption-cli.

Table of Contents

The TOC is grouped by purpose. Start at Diagnostic Workflows when triaging an active failure; the earlier sections are foundational references.

1. Fabric Foundations (concepts)

2. CLI Setup & Authentication

3. Spark Sessions, Notebooks & Jobs (background)

4. Spark Monitoring APIs (primary triage surface)

5. Diagnostic Workflows (start here for active triage)

6. Job Failure Diagnostics

7. Livy Session Health

8. Performance Diagnostics

9. Offline / Deep-Dive Tools

---

Must/Prefer/Avoid

MUST DO

• Always retrieve job/session status before attempting remediation
• Use workspace and item discovery from COMMON-CLI.md — never hardcode IDs
• Check Livy session state before submitting diagnostic statements
• Follow the Failure Triage Workflow for systematic diagnosis
• Always check the Spark Advisor API before reading raw logs — it often identifies the root cause immediately
• Use monitoring APIs (no active session required) before attempting Livy-based diagnostics
• Poll job/session status with 10–30 second intervals; timeout diagnostics after 30 minutes
• Always include the Notebook Snapshot URL in diagnostic output — it has the longest retention and enables cell-level inspection in the Fabric UI

PREFER

• Querying job instance history to establish baseline before declaring a regression
• Reusing existing idle sessions for diagnostic queries instead of creating new ones
• Checking capacity utilization when jobs are slow before blaming the Spark code
• Using az rest with JMESPath filtering to extract specific fields from large API responses
• The Spark Advisor API over manual log parsing for skew, task errors, and timeout detection
• Resource Usage API coreEfficiency metric to quantify cluster utilization before recommending scaling
• Job instance history comparison (last 5 runs) to detect regressions before deep-diving

AVOID

• Killing sessions without checking if they have active statements
• Creating new sessions for every diagnostic query (reuse idle sessions)
• Assuming OOM without checking actual memory metrics from Livy
• Hardcoded workspace or item IDs in diagnostic scripts
• Diagnosing performance without first checking capacity throttling via the Admin API
• Submitting diagnostic statements to sessions in busy state

---

Examples

Example 1: Diagnose a Failed Notebook

User prompt: "Why did my notebook ETL_Daily fail in workspace Production?"

Agent workflow: 1. Resolves workspace → workspaceId, item → itemId (Notebook) 2. Lists recent Livy sessions, auto-picks the Failed session 3. Queries Spark Advisor → finds TaskError: OutOfMemoryError on executor 4. Queries /stages → confirms data skew (12× max/median ratio in stage 5) 5. Presents report with HIGH findings + fix recommendations

Example 2: Triage Stuck Livy Session

User prompt: "My Livy session abc-1234 is stuck in starting state"

Agent workflow: 1. Uses session ID directly, queries session state 2. Lists all workspace sessions → detects 8 concurrent sessions (capacity pressure) 3. Checks Livy log → no errors, just queued 4. Reports: capacity contention, recommends waiting or cancelling idle sessions

Example 3: Pipeline Failure Root Cause

User prompt: "Diagnose pipeline run 5678 in workspace Analytics"

Agent workflow: 1. Resolves pipeline, calls queryActivityRuns for run 5678 2. Finds 2 Notebook activities: one Succeeded, one Failed 3. Extracts output.result.error.{ename, evalue, traceback} from failed activity 4. Constructs Notebook Snapshot URL for cell-level inspection 5. Presents error details + snapshot link + suggested fix

---

Quick Start

Environment Setup

Apply environment detection from COMMON-CLI.md to set:

• $FABRIC_API_BASE and $FABRIC_RESOURCE_SCOPE
• $FABRIC_API_URL and $LIVY_API_PATH for Livy operations

Authentication: Use token acquisition from COMMON-CLI.md § Authentication Recipes.

---

Automated Diagnostic Workflow

When the user provides a simple prompt (e.g., "Diagnose my notebook ETL_Pipeline", "What's wrong with Spark application abc-123", "Check workspace Production for issues"), follow this fast-path summary. For full procedure, edge cases (expired data, pipeline-only sessions), report templates, and retention details, see references/automated-diagnostic-workflow.md.

Entry Points (what the user provides)

Item-Type API Paths

All session API paths follow: $FABRIC_API_URL/workspaces/$workspaceId/<itemTypePath>/$itemId/livySessions/$livyId/applications/$appId/<endpoint> — see SPARK-MONITORING-CORE.md.

Steps at a Glance

Key principle: Always check Spark Advisor first — it's pre-computed and identifies most root causes without log parsing. Pipeline runs have the richest error data via queryActivityRuns (ename, evalue, traceback, cell/line) — see pipeline-diagnosis.md.

Data retention warning: Spark Monitoring API data (logs, stages, advisor) typically expires in minutes to hours after session end. Diagnose failures promptly. If APIs return 404, jump to Step 1b in the reference.

Tier 2 escalation: For truncated data, HTTP 408/504, or DAG/SQL plan visualization, suggest the offline Spark History Server workflow.

Automated Diagnostic Workflow — Reference

Companion to the compact summary in SKILL.md § Automated Diagnostic Workflow. This reference holds the verbose procedure, edge-case fallbacks, retention details, and report templates.

---

Step 1 — Resolve & Discover (full)

# Resolve workspace — list all matches, disambiguate if needed
workspaceMatches=$(az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces" \
  --query "value[?displayName=='<UserWorkspaceName>'].{id:id, displayName:displayName}" --output json)
matchCount=$(echo "$workspaceMatches" | jq length)
if [ "$matchCount" -eq 0 ]; then
  echo "ERROR: No workspace found with name '<UserWorkspaceName>'. Verify the name and retry."
  exit 1
elif [ "$matchCount" -gt 1 ]; then
  echo "Multiple workspaces match '<UserWorkspaceName>':"
  echo "$workspaceMatches" | jq -r '.[] | "  - \(.id)  \(.displayName)"'
  echo "Please specify the workspace ID directly."
  exit 1
fi
workspaceId=$(echo "$workspaceMatches" | jq -r '.[0].id')

# Resolve item (notebook, SJD, or lakehouse) — list all matches, disambiguate if needed
itemMatches=$(az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items?type=Notebook" \
  --query "value[?displayName=='<UserItemName>'].{id:id, displayName:displayName}" --output json)
itemCount=$(echo "$itemMatches" | jq length)
if [ "$itemCount" -eq 0 ]; then
  echo "Item '<UserItemName>' not found as Notebook. Trying SparkJobDefinition, then Lakehouse..."
  # Retry with ?type=SparkJobDefinition, then ?type=Lakehouse
elif [ "$itemCount" -gt 1 ]; then
  echo "Multiple items match '<UserItemName>':"
  echo "$itemMatches" | jq -r '.[] | "  - \(.id)  \(.displayName)"'
  echo "Please specify the item ID directly."
  exit 1
fi
itemId=$(echo "$itemMatches" | jq -r '.[0].id')

# List recent Livy sessions (sorted newest first)
# Use the correct item-type path:
#   /notebooks/{itemId}/livySessions
#   /sparkJobDefinitions/{itemId}/livySessions
#   /lakehouses/{itemId}/livySessions
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/<itemTypePath>/$itemId/livySessions" \
  --output json

Disambiguation: The resolve snippets above check match count before proceeding. If zero matches are found, an error is reported. If multiple workspaces or items share the same displayName, all matches are listed and the user is asked to specify the exact ID.

Item-type API paths:

Lakehouse note: Lakehouse Spark sessions are typically short-lived (table loads, maintenance). If livySessions returns empty, check jobs/instances for TableLoad/TableMaintenance job history. Lakehouse jobs do not have a Notebook Snapshot — use Spark Advisor and driver logs for diagnostics.

Present a session summary table to the user (most recent 10):

## Recent Sessions for <notebook name>

| # | Session ID | State | Submitted | Duration | App ID |
|---|------------|-------|-----------|----------|--------|
| 1 | abc-1234…  | Failed    | 2h ago  | 5m 23s  | app_…001 |
| 2 | def-5678…  | Succeeded | 4h ago  | 12m 10s | app_…002 |
| 3 | ghi-9012…  | Failed    | 1d ago  | 0s      | —        |

Session selection logic:

• Auto-pick if unambiguous — e.g., user said "why did it fail" and exactly 1 recent session has state == Failed → select it automatically and proceed
• Ask the user if ambiguous — multiple sessions match the user's intent (e.g., 2+ recent Failed sessions, or user said "diagnose" without specifying failed/slow) → present the table and ask which session to diagnose
• User provided session/app ID → skip the table entirely, use the ID directly

Extract livyId, sparkApplicationId, and state from the selected session.

---

Step 1b — Fallback: Session Not Found / Data Expired

If the user provided a Livy session ID but it is not found in any session listing (workspace-level or item-level) and Spark Monitoring APIs return 404:

Why this happens: Spark Monitoring API data (jobs, stages, executor logs, driver stderr) has limited retention after session completion — typically minutes to hours. Diagnose failures as soon as possible after they occur for the richest data.

1. Determine the notebook ID — ask the user if unknown:

I found no active data for session `<livyId>` via Spark Monitoring APIs (data retention expired).

To diagnose this session, I need the **notebook name or ID** it belongs to.
- If this was from a **pipeline run**, provide the pipeline name + run ID — `queryActivityRuns` may still have error details.
- If you know the **notebook name**, provide it and I'll construct a direct link to the Fabric UI snapshot.

2. Search pipeline runs (if user confirms pipeline origin or workspace has pipelines): Iterate pipelines → GET /items/$pipelineId/jobs/instances?limit=5 → for Failed runs, queryActivityRuns to find sessionId match. Returns output.result.error.{ename, evalue, traceback[]} — richest error data available.

3. Check Job Instance API — GET /items/$notebookId/jobs/instances?limit=5 for high-level failureReason (longer retention than Spark Monitoring APIs).

4. Construct Notebook Snapshot URL for manual cell-level inspection:

https://app.powerbi.com/workloads/de-ds/sparkmonitor/{notebookId}/{livyId}?trident=1&experience=power-bi&ctid={tenantId}&tab=related

The Fabric UI retains notebook snapshots much longer than Spark Monitoring APIs (shows failed cell, traceback, cell execution times, and source code).

5. Present report with all available data:

## Diagnostic Summary

**Session**: <livyId> | **Notebook**: <notebook name> | **State**: API data expired

### Error Details

[If queryActivityRuns returned data]:
**Exception**: <ename>: <evalue>
**Cell**: Cell In[<N>], line <M>
**Traceback**: <traceback lines>

[If only Job Instance data]:
**Failure Reason**: <failureReason from Job Instance API>

### Notebook Snapshot (cell-level details)
**Open Notebook Snapshot in Fabric UI**: `<constructed URL>`
↑ Click to view the exact failed cell, error output, and source code in the Fabric UI.

### Suggested Next Steps
1. Open the Notebook Snapshot link above to identify the exact failed cell and error
2. Fix the identified issue and re-run the notebook
3. For future failures, diagnose within 1 hour for full Spark Monitoring API data
4. For recurring failures, set up [proactive event log copy](jobinsight-api.md) to OneLake

Key principle: Exhaust all public APIs (queryActivityRuns → Job Instance → Spark Monitoring) before falling back to the manual Notebook Snapshot URL. Always present the snapshot link — it has the longest retention.

---

Step 2 — Auto-Route by Session State

---

Step 3 — Failure Analysis

Error API priority — query in this order, stop when root cause is clear: 1. Spark Advisor (/advice) — automated root-cause with fix recommendations 2. Driver stderr (/logs?type=driver&fileName=stderr&isDownload=true) — raw exception stack traces 3. Job Instance (/jobs/instances/{id}) — high-level failureReason 4. Executor logs (/logs?type=executor&meta=true) — per-executor OOM / ExecutorLostFailure 5. Livy log (/logs?type=livy) — startup errors, library packaging failures 6. Resource Usage (/resourceUsage) — capacityExceeded, task limit exhaustion 7. Notebook Snapshot URL (manual) — all APIs expired, see Step 1b

For pipeline runs, queryActivityRuns (Step P2 in pipeline-diagnosis.md) is the richest single source — returns output.result.error.{ename, evalue, traceback[]} with cell/line numbers.

All API paths follow the pattern: $FABRIC_API_URL/workspaces/$workspaceId/<itemTypePath>/$itemId/livySessions/$livyId/applications/$appId/<endpoint> — see SPARK-MONITORING-CORE.md for full specs.

Auto-classify errors by matching log content against the Quick Reference Table.

---

Step 4 — Performance Analysis

Query /stages and /allexecutors endpoints (see SPARK-MONITORING-CORE.md § Open-Source Spark History Server APIs).

Auto-flag using Detection Thresholds:

• Data skew: max/median task duration > 3×
• Disk spill: diskBytesSpilled > 0
• GC pressure: jvmGcTime/executorRunTime > 20%
• Heavy shuffle: shuffleWriteBytes > 1 GB
• Small partitions: high task count, < 100 ms each

---

Step 5 — Resource Utilization

Query /resourceUsage endpoint (see SPARK-MONITORING-CORE.md § Resource Usage API). Extract coreEfficiency, idleTime, duration.

Auto-flag:

• coreEfficiency < 0.3 → HIGH (underutilized)
• idleTime / duration > 0.4 → MEDIUM (high idle)

---

Step 6 — Session Health

List all sessions via GET /workspaces/$workspaceId/spark/livySessions. Auto-flag:

• idle with no recent statements → zombie
• starting beyond expected duration → capacity issue
• many concurrent sessions → capacity pressure

---

Step 7 — Compile & Present Report

After running the applicable steps, present a structured report:

## Diagnostic Summary

**Application**: <notebook name> | **Session**: <livyId> | **State**: <state>

### Findings (ordered by severity)

| # | Severity | Category | Finding | Recommended Fix |
|---|----------|----------|---------|-----------------|
| 1 | HIGH     | Failure  | Driver OOM from collect() on line 45 | Replace with df.write.parquet() |
| 2 | HIGH     | Perf     | Data skew in stage 12 (8.2× ratio) | Enable AQE skew join |
| 3 | MEDIUM   | Perf     | Disk spill in stage 8 (2.1 GB) | Increase shuffle partitions |
| 4 | MEDIUM   | Resource | Core efficiency 22% | Reduce executor count |

### Links
- **Notebook Snapshot**: `https://app.powerbi.com/workloads/de-ds/sparkmonitor/{notebookId}/{livyId}?trident=1&experience=power-bi&ctid={tenantId}&tab=related`
- **Spark Monitor**: `https://app.powerbi.com/workloads/de-ds/sparkmonitor/{notebookId}/{livyId}?trident=1&experience=power-bi&ctid={tenantId}`

### Suggested Next Steps
1. [Most impactful fix first]
2. [Second fix]
3. [Optional: escalate to Tier 2 if needed]

Notebook Snapshot URL host: Use app.powerbi.com for production, msit.powerbi.com for MSIT.

Tier 2 escalation: If any step returns truncated data, HTTP 408/504, or the user asks for DAG/SQL plan visualization, suggest the offline workflow.

---

Data Retention Summary

Public API retention windows for diagnostics:

Implication: Diagnose failures as soon as possible. For recurring failures, configure proactive event log copy to OneLake for permanent retention.

Diagnostic Workflow Guide

Manual CLI Recipes

The following recipes are for ad-hoc manual use. The automated workflow is preferred for most users.

Diagnose a Failed Notebook Run

# 1. Discover workspace and notebook
workspaceId=$(az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces" \
  --query "value[?displayName=='MyWorkspace'].id" --output tsv)

notebookId=$(az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items?type=Notebook" \
  --query "value[?displayName=='MyNotebook'].id" --output tsv)

# 2. Get recent job instances
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances?limit=5" \
  --query "value[].{id:id, status:status, start:startTimeUtc, end:endTimeUtc, failureReason:failureReason}" \
  --output table

# 3. Get details of the failed instance
jobInstanceId="<from-above>"
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances/$jobInstanceId" \
  --output json

Check Livy Session Health

# List all sessions for a lakehouse
lakehouseId="<lakehouse-id>"
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions" \
  --query "sessions[].{id:id, state:state, name:name, appId:appId}" \
  --output table

# Get detailed session info (includes memory, executors)
sessionId="<session-id>"
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId" \
  --output json

Quick Performance Check via Livy Statement

# Run a diagnostic PySpark snippet in an existing idle session
cat > /tmp/body.json << 'DIAG'
{
  "code": "sc = spark.sparkContext\nprint('Active executors:', len(sc._jsc.sc().getExecutorMemoryStatus()))\nprint('Default parallelism:', sc.defaultParallelism)\nprint('Spark config:')\nfor k, v in sorted(spark.sparkContext.getConf().getAll()):\n    if any(x in k for x in ['memory', 'cores', 'parallelism', 'shuffle', 'dynamic']):\n        print(f'  {k} = {v}')",
  "kind": "pyspark"
}
DIAG
az rest --method post --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/statements" \
  --body @/tmp/body.json --output json

---

Key Diagnostic Patterns

---

Diagnostic Tiers

This skill provides two diagnostic tiers. Always start with Tier 1. Escalate to Tier 2 only when Tier 1 is insufficient.

Tier 1 — Online (Primary)

Uses Fabric Spark Monitoring REST APIs (via az rest) to pull session data, failed jobs, slowest stages, Spark Advisor findings, and driver logs. Fast; no download; no active session required.

Workflow: 1. Find the session — List Livy sessions for the notebook/SJD (see Monitoring Workflow) 2. Check Advisor — Query the Spark Advisor API for automated root-cause detection 3. Failure analysis — If job failed, inspect failed jobs/stages, read driver/executor logs 4. Performance analysis — Check stage metrics, executor utilization, resource usage 5. Interpret and report — Classify findings using the severity thresholds below

Tier 2 — Offline Fallback (Local Spark History Server)

Copies the full Spark event log from Fabric to a OneLake lakehouse via the JobInsight API, downloads it locally, and starts the OSS Spark History Server on http://localhost:18080 for the full Spark UI (DAG, task-level detail, SQL plan visualizations).

When to escalate to Tier 2:

Offline workflow: 1. Copy event logs via JobInsight LogUtils.copyEventLog() — see jobinsight-api.md 2. Download event files from OneLake DFS to local disk 3. Start local Spark History Server — see spark-history-server.md 4. Open browser to http://localhost:18080 for the full Spark UI

---

Severity Thresholds

Use these thresholds when interpreting stage metrics and resource usage to classify issues:

Stage Detail Selection Logic

When analyzing performance, select stages for detailed task-level inspection: 1. If Spark Advisor identifies affected stages/jobs/SQL executions — use those first 2. If Advisor has no findings, select fallback candidates: slowest 10 stages, shuffle-heavy top 10, spill top 10 3. Union candidates and cap at 20 stages total 4. For each selected stage, query /stages/{stageId}/{attemptId}/taskSummary?quantiles=0.25,0.5,0.75,0.95,1.0

Presenting Results

1. Lead with a one-line severity summary (e.g., "Found 2 job failures and 3 performance issues") 2. Classify each finding by severity (HIGH/MEDIUM/LOW) 3. Highlight the most impactful fix first 4. Offer to drill into any specific failure or bottleneck

Job Diagnostics

Scope: Classify Spark job failures, retrieve logs via Fabric REST APIs, analyze job instance history, and follow a systematic triage workflow. All examples use az rest against the Fabric API.

---

Failure Classification

Spark job failures in Fabric fall into distinct categories. Identify the category first — it determines the remediation path.

1. Out of Memory — Driver

Signature: java.lang.OutOfMemoryError: Java heap space in driver stderr Also matches: OutOfMemoryError: GC overhead limit exceeded

Root Cause: The Spark Driver JVM ran out of heap memory. Most commonly caused by collecting large amounts of data to the driver with collect(), toPandas(), or show() on a large DataFrame.

Common Causes:

Fix:

# Instead of collect: write to storage
df.write.mode("overwrite").parquet("Files/output/")

# Instead of toPandas on large data: sample first
df.sample(0.01).toPandas()

# Instead of show: use display() which is paginated
display(df)

# If you must collect: always check count first
count = df.count()
if count < 100_000:
    df.collect()
else:
    raise ValueError(f"Too many rows to collect: {count}")

Spark Config Fix (if driver must process large data):

spark.conf.set("spark.driver.memory", "8g")  # set in notebook config, not at runtime

2. Out of Memory — Executor

Signature: java.lang.OutOfMemoryError in executor stderr Also matches: Container killed by YARN for exceeding memory limits, ExecutorLostFailure (executor N exited caused by one of the running tasks)

Root Cause: An executor exceeded its JVM heap + overhead memory budget. Caused by large shuffle aggregations, wide transformations with many columns, or UDFs that create large intermediate objects.

Common Causes:

Fix:

# Increase executor memory in Fabric notebook Spark settings (spark pool config)
# spark.executor.memory = 4g (default varies by pool size)

# Reduce data per partition
df = df.repartition(200)

# Avoid collecting broadcast data larger than spark.broadcast.blockSize
# Use SortMergeJoin instead of BroadcastHashJoin for large tables
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")  # disable auto-broadcast

# Enable off-heap memory for Tungsten
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "2g")

Diagnostic Query (run in Livy session):

# Check for skewed partitions (large variance = skew)
df = spark.table("your_table")
partition_sizes = df.groupBy(spark.sparkContext.partitionId().alias("pid")).count()
partition_sizes.describe("count").show()
# If max >> mean, you have data skew

3. Shuffle Fetch Failed

Signature: org.apache.spark.shuffle.FetchFailedException Also matches: Failed to get broadcast_N, ShuffleMapTask failed with FetchFailed

Root Cause: An executor tried to fetch shuffle data (map output) from another executor, but that executor died or the shuffle block was lost. Often a secondary symptom of OOM killing the producer executor.

Common Causes:

Fix:

# Increase executor memory to prevent the producer from dying
# Enable AQE to handle skew that might overload one executor
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Reduce shuffle data by filtering early
df = df.filter(col("date") >= "2024-01-01")  # push filter before shuffle

# Increase shuffle retry tolerance
spark.conf.set("spark.shuffle.maxRetriesOnNetIssue", "10")
spark.conf.set("spark.shuffle.retryWait", "5s")

4. Executor Lost

Signature: ExecutorLostFailure (executor N exited caused by one of the running tasks) Also matches: Lost executor N on host

Root Cause: The executor process was killed by the OS or resource manager, typically due to memory overuse (container eviction) or a JVM crash.

Fix:

• Check if OOM errors precede this in the log (see pattern #2)
• Reduce memory pressure: smaller partitions, less data per task
• Enable speculative execution to tolerate slow executors:

spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", "1.5")

5. Analysis Exception (SQL / Schema Error)

Signature: org.apache.spark.sql.AnalysisException Also matches: cannot resolve column, cannot up cast, Column 'X' does not exist

Root Cause: Spark cannot resolve a column name, function, or data type at query planning time. Caused by typos in column names, schema mismatches, or using a column that was dropped/renamed earlier.

Fix:

# Print schema to verify column names
df.printSchema()

# Check available columns
print(df.columns)

# Use backticks for column names with spaces or special chars
df.select("`my column`")

# Verify the column exists before using it
assert "user_id" in df.columns, "user_id column missing"

6. File Not Found / Path Error

Signature: java.io.FileNotFoundException Also matches: Path does not exist, No such file or directory

Root Cause: A file or directory path referenced in the code does not exist in the Lakehouse or ABFSS storage. Common causes: wrong path prefix, file was deleted, or a previous write step failed.

Fix:

# Use the correct Fabric Lakehouse path format
df = spark.read.parquet("abfss://<workspace>@<storage>.dfs.core.windows.net/<path>")

# Or use the Files shortcut in Fabric notebooks
df = spark.read.parquet("Files/my_folder/my_file.parquet")

# Check existence before reading
from pyspark.sql import SparkSession
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
path = spark._jvm.org.apache.hadoop.fs.Path("Files/my_folder")
if not fs.exists(path):
    raise FileNotFoundError(f"Path does not exist: {path}")

7. Task Killed — Speculative Execution (Informational)

Signature: TaskKilled (another attempt succeeded)

Root Cause: This is normal behavior from Spark's speculative execution. When a task runs slower than peers, Spark launches a duplicate. The slower duplicate is killed when the faster one completes.

Action: None required. If these appear frequently (many tasks killed per stage), the cluster may be over-provisioned or data is skewed — see performance-patterns.md for skew detection.

8. Library Packaging Error

Signature: library packaging error in Livy log Also matches: Failed to install, pip install failed

Root Cause: A custom Python library or wheel file specified in the session environment failed to install — usually due to a version conflict, missing dependency, or network issue.

Fix: 1. Verify the library version exists for the Fabric runtime's Python version 2. Check for conflicting dependencies:

   pip install <lib> --dry-run

3. Upload the wheel file directly to Lakehouse Files and install inline:

   %pip install /lakehouse/default/Files/mylib-1.0-py3-none-any.whl

9. Spark SQL Parse Exception

Signature: org.apache.spark.sql.catalyst.parser.ParseException

Root Cause: Invalid SQL syntax in a spark.sql("...") call or SQL magic cell.

Fix:

# Use triple quotes and test the SQL separately
query = """
    SELECT
        user_id,
        COUNT(*) AS event_count
    FROM events
    WHERE date >= '2024-01-01'
    GROUP BY user_id
"""
spark.sql(query).show(5)

10. Broadcast Timeout

Signature: SparkException: Could not execute broadcast in N secs Also matches: org.apache.spark.SparkException: Broadcast timeout

Root Cause: The driver took too long to broadcast a table to all executors, usually because the "small" table is actually too large for broadcast.

Fix:

# Increase timeout (default 300s)
spark.conf.set("spark.sql.broadcastTimeout", "600")

# Or disable auto-broadcast and switch to SortMergeJoin
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

# Or explicitly hint the join type instead
from pyspark.sql.functions import broadcast
df_large.join(broadcast(df_truly_small), "key")

Timeout Failures

Symptoms:

• Job cancelled because SparkContext was shut down
• Task not serializable
• Session transitions to dead state after long idle
• Livy session has expired

Common Causes:

---

Quick Reference Table

---

Reading Spark Logs via REST

Tip: For completed or failed Spark applications, the Driver and Executor Log APIs provide direct REST access to logs without requiring an active Livy session. The Livy Log API offers byte-offset pagination for large logs.

Retrieve Livy Session Logs

Livy exposes driver logs through the session API. This is the primary way to get Spark logs in Fabric without accessing the cluster directly.

# Get the last 100 lines of driver output
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/log?from=0&size=100" \
  --query "log" --output tsv

Pagination for large logs:

# Get total log lines first
total=$(az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/log?from=0&size=1" \
  --query "total" --output tsv)

# Get the last 200 lines (where errors usually are)
from=$((total - 200))
[ $from -lt 0 ] && from=0
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/log?from=$from&size=200" \
  --query "log" --output tsv

Retrieve Statement Output for Errors

When a Livy statement fails, the error is in the statement output:

# Get statement result (includes traceback for failed statements)
statementId="<statement-id>"
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/statements/$statementId" \
  --query "{state:state, output:output}" --output json

The output object contains:

• status: "ok" or "error"
• evalue: Error message text
• traceback: Full Python/Java traceback as array of strings

---

Job Instance History

Query Recent Job Runs

Use job instance APIs to compare runs over time and detect regressions.

# Get last 10 job instances for a notebook
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances?limit=10" \
  --query "value[].{id:id, status:status, start:startTimeUtc, end:endTimeUtc, failureReason:failureReason}" \
  --output table

Compare Job Durations

# Extract durations for trend analysis
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances?limit=20" \
  --query "value[?status=='Completed'].{start:startTimeUtc, end:endTimeUtc}" \
  --output json

To compute duration differences, pipe through jq or process in a Livy session:

# Using jq to compute durations (if available)
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances?limit=20" \
  --output json | jq '.value[] | select(.status=="Completed") |
    {start: .startTimeUtc, end: .endTimeUtc, status: .status}'

Detect Regressions

A job is regressing if the latest successful run is significantly slower than the median of the previous runs. Compare the last run's duration against the median of the 5 runs before it. If the ratio exceeds 2x, investigate data volume changes first, then Spark configuration drift.

---

Failure Triage Workflow

Follow this decision tree when a Spark job fails in Fabric.

Step 1: Get the Job Status

# What is the job's current state?
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances/$jobInstanceId" \
  --query "{status:status, failureReason:failureReason}" --output json

Step 2: Classify the Failure

Read failureReason from the job instance response. Match against the categories in Failure Classification:

1. Contains `OutOfMemoryError` → OOM category 2. Contains `FetchFailedException` or `ShuffleMapTask` → Shuffle failure 3. Contains `timeout` or `expired` → Timeout category 4. Contains `ClassNotFoundException` or `AnalysisException` → Dependency/config error 5. None of the above → Read the full Livy session log (Step 4)

Step 3: Check for Cancellation Cause

# Was the job cancelled by the user or by the system?
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances/$jobInstanceId" \
  --query "{status:status, invokeType:invokeType, failureReason:failureReason}" --output json

• invokeType: "Manual" + cancelled → user cancelled
• System cancellation → check capacity throttling or session timeout

Step 4: Read the Logs

If failureReason is not descriptive enough, retrieve the Livy session logs using the patterns in Reading Spark Logs via REST. Search for:

• ERROR or FATAL log lines
• Java exception stack traces (lines starting with at )
• Python tracebacks (lines starting with Traceback or File ")

Alternative: Use the Driver Log API to access driver stderr directly, or the Executor Log API for per-executor logs. These APIs work on completed applications without an active session.

Step 4b: Check Spark Advisor

Before manual analysis, check if the Spark Advisor has already identified the issue:

az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/notebooks/$notebookId/livySessions/$livyId/applications/$appId/advice" \
  --output json

The Advisor automatically detects data skew, time skew, and task errors. See Spark Advisor API.

Step 5: Apply the Fix

Once classified, apply the remediation from the corresponding category in Failure Classification. After applying, re-run the job and monitor using the Job Instance History patterns to confirm the fix.

JobInsight API Reference

Scope: Copy Spark event logs from Fabric to OneLake for offline analysis using the JobInsight Scala library. This enables Tier 2 diagnostics (local Spark History Server) when online APIs are insufficient.

---

Overview

JobInsight is a Scala library available on Fabric Spark clusters that provides diagnostic utilities for Spark job analysis. It is not available as a standalone package — it must run inside a Fabric Spark notebook or Livy session.

---

LogUtils.copyEventLog

import com.microsoft.jobinsight.diagnostic.LogUtils

val contentLength: Long = LogUtils.copyEventLog(
    workspaceId: String,    // Fabric workspace UUID
    artifactId: String,     // Notebook / artifact UUID
    livyId: String,         // Spark Livy session UUID
    jobType: String,        // "sessions", "batches", etc.
    targetDir: String,      // abfss:// path where event logs are written
    overwrite: Boolean,     // true to overwrite existing logs (default: true)
    attemptId: Integer      // YARN attempt number; null = auto-detect, 1 = first attempt
)

Parameters

Return Value

Returns a Long representing the total number of bytes copied.

Target Directory Format

The recommended target directory pattern for lakehouse storage:

abfss://{workspaceId}@onelake.dfs.fabric.microsoft.com/{lakehouseId}/Files/spark-events/{livyId}/

This places event logs under the lakehouse Files section, organized by session.

---

Requirements

---

Attempt ID (7th Parameter)

The attemptId parameter controls how copyEventLog matches event-log files in blob storage.

Internal flow: 1. CredentialProvider.getEventLogDirSasInfo() → gets a SAS-scoped directory URL 2. CredentialProvider.listFiles() → lists blobs under that directory 3. CredentialProvider.eventLogNamePattern(attemptId) → builds a regex to filter files 4. Files matching the regex are copied to targetDir

Regex behavior:

Known issue: Spark event-log files almost always include an attempt suffix (e.g. _1). When attemptId = null, the regex uses String.matches() (full match) and does not match file names with the _1 suffix. Auto-detection queries cluster metadata to resolve the attempt number, but this metadata is garbage-collected within ~3 days of session completion. After that, the call falls back to the null regex and silently fails.

Recommendation: Always pass attemptId = 1 unless you know the session had multiple YARN attempts (very rare for interactive notebook sessions).

---

Common Errors

---

Usage Example

Direct in Fabric Notebook (Scala cell)

import com.microsoft.jobinsight.diagnostic.LogUtils

val workspaceId = "4cb9b656-c8f8-485e-a151-e81bb913abc8"
val notebookId  = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
val livyId      = "12345678-1234-1234-1234-123456789abc"
val lakehouseId = "abcdef12-3456-7890-abcd-ef1234567890"

val targetDir = s"abfss://$workspaceId@onelake.dfs.fabric.microsoft.com/$lakehouseId/Files/spark-events/$livyId/"

val bytesWritten = LogUtils.copyEventLog(
    workspaceId, notebookId, livyId,
    "sessions", targetDir, true, 1
)
println(s"Copied $bytesWritten bytes of event logs")

Via Notebook Run API (Automation Pattern)

Since JobInsight is Scala-only and cluster-bound, the recommended automation pattern is:

1. Create a parameterized Scala notebook using a Fabric parameter cell (toggle the cell to "Parameter" in the notebook toolbar) to declare default values for workspaceId, livyId, and targetDir 2. Run it via the Fabric Notebook Run API:

   az rest --method post --resource "https://api.fabric.microsoft.com" \
     --url "https://api.fabric.microsoft.com/v1/workspaces/$workspaceId/items/$notebookId/jobs/RunNotebook/instances" \
     --body '{"executionData": {"parameters": {"workspaceId": "...", "livyId": "...", "targetDir": "..."}}}'

3. Poll the Location header for completion (see COMMON-CLI.md § LRO Pattern)

---

Downloading Event Logs from OneLake

After copyEventLog writes event logs to OneLake, download them locally for the Spark History Server:

# Get a storage token
TOKEN=$(az account get-access-token --resource https://storage.azure.com --query accessToken -o tsv)

# List event log files
curl -s -H "Authorization: Bearer $TOKEN" \
  "https://onelake.dfs.fabric.microsoft.com/$workspaceId/$lakehouseId/Files/spark-events/$livyId?resource=filesystem&recursive=true" \
  | jq -r '.paths[].name'

# Download each file to local directory
LOCAL_DIR="$HOME/.spark-local/event-logs/$livyId"
mkdir -p "$LOCAL_DIR"

# For each file listed above:
curl -s -H "Authorization: Bearer $TOKEN" \
  "https://onelake.dfs.fabric.microsoft.com/$workspaceId/$lakehouseId/<filepath>" \
  -o "$LOCAL_DIR/<filename>"

Then start the local Spark History Server — see spark-history-server.md.

Performance Patterns

Scope: Identify and resolve Spark performance bottlenecks in Microsoft Fabric — covering detection thresholds, anti-patterns, stage/task analysis, optimization recipes, and capacity diagnostics. All examples use az rest and PySpark via Livy sessions.

---

Detection Thresholds

The diagnostic skill uses the following thresholds to flag performance issues:

---

Anti-Patterns

These are the most common performance killers in Fabric Spark workloads. Each pattern includes detection methods and fixes.

Shuffle Spill

What it is: When shuffle data exceeds available memory, Spark spills data to disk. Disk I/O is orders of magnitude slower than memory access.

Detection (via Livy PySpark statement):

# Check for spill in the last completed job
from pyspark.sql import SparkSession
sc = spark.sparkContext

# Get status URL info
print("Spark UI available via Fabric Monitoring Hub")
print(f"Application ID: {sc.applicationId}")

# Check current Spark config for shuffle settings
for key in ['spark.sql.shuffle.partitions', 'spark.shuffle.spill.compress',
            'spark.sql.adaptive.enabled', 'spark.sql.adaptive.coalescePartitions.enabled']:
    print(f"{key} = {sc.getConf().get(key, 'not set')}")

Fixes:

Data Skew

What it is: Uneven distribution of data across partitions. A few partitions have vastly more data than others, causing some tasks to run much longer.

Detection:

# Detect skew in a DataFrame
df = spark.table("your_table")

# Check partition sizes
from pyspark.sql.functions import spark_partition_id, count
partition_stats = df.groupBy(spark_partition_id().alias("partition")) \
    .agg(count("*").alias("row_count"))

stats = partition_stats.describe("row_count")
stats.show()

# Skew ratio: max / mean — values > 3x indicate significant skew
max_count = partition_stats.agg({"row_count": "max"}).collect()[0][0]
mean_count = partition_stats.agg({"row_count": "avg"}).collect()[0][0]
print(f"Skew ratio: {max_count / mean_count:.1f}x (>3x = significant skew)")

Fixes:

Small File Problem

What it is: Too many small files in Delta tables cause excessive metadata overhead and slow reads.

Detection:

# Check file sizes in a Delta table
from delta.tables import DeltaTable
dt = DeltaTable.forName(spark, "your_table")

# File count and sizes
files_df = spark.sql("DESCRIBE DETAIL your_table")
files_df.select("numFiles", "sizeInBytes").show()

# Rule of thumb: aim for 128MB-1GB per file
size_bytes = files_df.collect()[0]["sizeInBytes"]
num_files = files_df.collect()[0]["numFiles"]
avg_file_mb = (size_bytes / num_files) / (1024 * 1024) if num_files > 0 else 0
print(f"Average file size: {avg_file_mb:.1f} MB")
if avg_file_mb < 32:
    print("⚠️  Small file problem detected — consider OPTIMIZE")

Fix:

# Compact small files
spark.sql("OPTIMIZE your_table")

# For partitioned tables, optimize specific partitions
spark.sql("OPTIMIZE your_table WHERE date = '2024-01-15'")

# Enable auto-optimize for future writes
spark.sql("ALTER TABLE your_table SET TBLPROPERTIES ('delta.autoOptimize.optimizeWrite' = 'true')")

Collect Misuse

What it is: Calling .collect(), .toPandas(), or .show(n) with large n on a large DataFrame pulls all data to the driver, causing OOM or extreme slowness.

Detection: Look for these patterns in notebook code:

• df.collect() on DataFrames with > 10K rows
• df.toPandas() without prior .limit()
• df.show(1000000) or similar large show calls

Fixes:

Cartesian Joins

What it is: A join without a proper join condition produces the Cartesian product — row count = left × right.

Detection: Check for cross joins in the query plan:

df_result = df1.join(df2, ...)  # suspect join
df_result.explain(True)  # look for "CartesianProduct" or "BroadcastNestedLoopJoin"

Fix: Always specify explicit join conditions. If a cross join is intended, use df1.crossJoin(df2) to make intent clear and add a downstream filter.

High GC Pressure

What it is: The JVM heap is filling up faster than garbage collection can free it. Executor CPU time is dominated by GC; throughput is much lower than raw CPU capacity suggests.

Detection: gcTime / executorRunTime > 20%

Root Cause:

• Python UDFs that create Python objects in every row
• String-heavy DataFrames with high cardinality
• Many small objects created by complex aggregation logic

Fix:

# Option 1: Replace Python UDFs with native PySpark SQL functions
# BAD — UDF creates Python objects per row
from pyspark.sql.functions import udf
@udf("string")
def clean(s):
    return s.strip().lower() if s else ""

# GOOD — uses JVM-native function, no Python object creation
from pyspark.sql.functions import lower, trim
df = df.withColumn("cleaned", lower(trim(col("value"))))

# Option 2: Use Pandas UDFs (vectorized) when Python logic is mandatory
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf("double")
def compute_score(values: pd.Series) -> pd.Series:
    return values.apply(lambda x: x * 2.5)

# Option 3: Cache DataFrames that are used multiple times
df_expensive.cache()
df_expensive.count()  # trigger materialization

# Option 4: Use Parquet/Delta format (columnar, minimizes Java object count)
df.write.format("delta").save("Files/output/")

Heavy Shuffle

What it is: Too much data is being shuffled across the network. Stage takes a long time; many "shuffle write" / "shuffle read" bytes visible in stage metrics.

Detection: shuffleWriteBytes > 1 GB in a stage

Root Cause:

• groupBy + aggregation on a high-cardinality column
• Multiple chained joins without caching intermediate results
• repartition(N) with N larger than needed

Fix:

# Option 1: Cache intermediate DataFrames to avoid re-shuffling
df_joined = large1.join(large2, "key").cache()
df_joined.count()  # materialize
result1 = df_joined.groupBy("category").agg(...)
result2 = df_joined.filter(col("status") == "active")

# Option 2: Reduce join width — select only needed columns before join
df_slim = df.select("key", "value1", "value2")
df_slim.join(other_slim, "key")

# Option 3: Use bucket tables for repeated large-table joins (eliminates shuffle)
df.write.bucketBy(64, "user_id").sortBy("user_id").saveAsTable("events_bucketed")
events = spark.table("events_bucketed")
users  = spark.table("users_bucketed")
events.join(users, "user_id")  # no shuffle!

# Option 4: Tune shuffle partitions to match data volume
# Rule of thumb: each shuffle partition should be 100-200 MB
spark.conf.set("spark.sql.shuffle.partitions", "100")
# Or let AQE coalesce automatically:
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

Too Many Small Partitions

What it is: High number of tasks but very short duration per task (< 100ms). Task scheduling overhead dominates actual compute time.

Detection: Stage shows 1000+ tasks but each completes in milliseconds.

Root Cause: spark.sql.shuffle.partitions is set too high relative to data volume, or source files are very small (many small Parquet files).

Fix:

# Reduce shuffle partitions globally
spark.conf.set("spark.sql.shuffle.partitions", "50")  # tune to your data size

# Or let AQE coalesce small partitions automatically (recommended)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

# For source files: coalesce small files before processing
df = spark.read.parquet("Files/raw/")
df = df.coalesce(20)  # reduce from 500 tiny partitions to 20 larger ones

Driver Memory Bottleneck

What it is: Operations that run on the driver rather than executors. All executors idle while driver is computing.

Detection: Driver OOM (see job-diagnostics.md pattern #1) or very long "driver computation" phases with idle executors.

Root Cause:

• df.collect() — materializes entire DataFrame in driver memory
• df.toPandas() — same
• Schema inference on many files
• Building broadcast variables from large datasets

Fix:

# Instead of collect(): write results
df.write.mode("overwrite").parquet("Files/output/")
display(df.limit(1000))  # paginate for visual inspection

# For schema inference on many files: provide schema explicitly
from pyspark.sql.types import StructType, StructField, StringType, LongType
schema = StructType([
    StructField("id", LongType(), True),
    StructField("name", StringType(), True),
])
df = spark.read.schema(schema).json("Files/data/")

Missing Cache / Re-computation

What it is: A DataFrame that is used multiple times is not cached, so Spark re-executes the full lineage from the source each time an action is called. Same stage appears multiple times with identical input/output.

Detection: Multiple downstream actions re-trigger the same expensive computation (visible in stage repetition).

Fix:

# BEFORE (expensive re-computation)
df_expensive = df.join(other, "key").groupBy("cat").agg(...)
result1 = df_expensive.filter(col("cat") == "A").count()
result2 = df_expensive.filter(col("cat") == "B").show()
# Both calls re-execute the join and aggregation

# AFTER (cache and reuse)
df_expensive = df.join(other, "key").groupBy("cat").agg(...).cache()
df_expensive.count()  # trigger materialisation

result1 = df_expensive.filter(col("cat") == "A").count()
result2 = df_expensive.filter(col("cat") == "B").show()

# ALWAYS unpersist when done
df_expensive.unpersist()

---

Stage and Task Analysis

Reading Spark Execution Metrics

In Fabric, the Spark UI is accessible through the Monitoring Hub. For programmatic access, use Livy to query runtime metrics.

# Get active and completed job info
sc = spark.sparkContext
status_tracker = sc.statusTracker()

# Active jobs
active_jobs = status_tracker.getActiveJobIds()
print(f"Active jobs: {list(active_jobs)}")

# For each active job, get stage info
for job_id in active_jobs:
    job_info = status_tracker.getJobInfo(job_id)
    print(f"Job {job_id}: status={job_info.status()}, "
          f"stages={list(job_info.stageIds())}")

Identify Slow Stages

# Run a query and then analyze the plan
df = spark.sql("SELECT ... FROM ... JOIN ...")

# Physical plan shows shuffle boundaries and join strategies
df.explain("formatted")

# The plan reveals:
# - Exchange nodes = shuffle boundaries (expensive)
# - BroadcastHashJoin vs SortMergeJoin (broadcast = faster for small tables)
# - Filter pushdown (filters should appear as early as possible)

Monitor Active Statements

# Check all statements in a session
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/statements" \
  --query "statements[].{id:id, state:state, progress:progress}" --output table

---

Optimization Recipes

Partition Tuning

# Check current partition count
df = spark.table("your_table")
print(f"Current partitions: {df.rdd.getNumPartitions()}")

# Rule of thumb: 128MB per partition, 2-4x parallelism of cluster
# For a 10GB dataset on 8 cores: ~80-320 partitions
optimal = max(1, int(df.inputFiles().__len__() * 128 / 1024))  # rough estimate
print(f"Suggested partitions: {optimal}")

# Repartition if needed
df_optimized = df.repartition(optimal)

Broadcast Join Optimization

from pyspark.sql.functions import broadcast

# For small dimension tables (< 100MB), broadcast to avoid shuffle
dim_df = spark.table("small_dimension_table")
fact_df = spark.table("large_fact_table")

# Explicit broadcast hint
result = fact_df.join(broadcast(dim_df), "key_column")

# Check if broadcast was used
result.explain()  # Should show BroadcastHashJoin, not SortMergeJoin

Caching Strategy

# Cache intermediate results that are reused multiple times
intermediate_df = spark.sql("""
    SELECT customer_id, SUM(amount) as total
    FROM transactions
    GROUP BY customer_id
""")

# Cache only if the DataFrame is reused in multiple downstream operations
intermediate_df.cache()
intermediate_df.count()  # Materialize the cache

# Use in multiple downstream queries
top_customers = intermediate_df.filter("total > 10000")
segments = intermediate_df.join(dim_customers, "customer_id")

# ALWAYS unpersist when done
intermediate_df.unpersist()

AQE (Adaptive Query Execution) Settings

AQE is enabled by default in Fabric. Verify and tune:

# Check AQE settings
aqe_keys = [
    'spark.sql.adaptive.enabled',
    'spark.sql.adaptive.coalescePartitions.enabled',
    'spark.sql.adaptive.skewJoin.enabled',
    'spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes',
    'spark.sql.adaptive.advisoryPartitionSizeInBytes'
]
for key in aqe_keys:
    print(f"{key} = {spark.conf.get(key, 'not set')}")

Predicate Pushdown Verification

# Verify that filters are pushed down to the data source
df = spark.table("your_table").filter("date >= '2024-01-01'").filter("status = 'active'")

# Check the physical plan — filters should appear at the scan level
df.explain(True)
# Look for: PushedFilters: [IsNotNull(date), GreaterThanOrEqual(date,2024-01-01)]
# If filters appear ABOVE the scan, pushdown is not working

---

Capacity and Resource Diagnostics

Check Capacity Utilization

When jobs are slow across the board, the issue may be capacity-level, not job-level.

Per-Application Resource Metrics: Use the Resource Usage API for granular vCore allocation and utilization timelines per Spark application, including core efficiency and idle time metrics.

# List available capacities and their state
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/capacities" \
  --query "value[].{name:displayName, id:id, sku:sku, state:state}" \
  --output table

Detect Throttling

Fabric throttles workloads when capacity utilization exceeds limits. Symptoms:

• Jobs take longer than usual to start
• Sessions stay in starting state
• API calls return 429 (Too Many Requests)

# Check if any recent API calls were throttled (429 responses)
# This is visible in job instance details
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$notebookId/jobs/instances?limit=5" \
  --query "value[].{status:status, start:startTimeUtc, failureReason:failureReason}" \
  --output table

Capacity Sizing Guide

Concurrent session counts are approximate and depend on session size (memory/cores). Starter Pool sessions consume fewer CUs than custom pool sessions.

Resource Efficiency Check

Run this diagnostic in a Livy session to check for resource waste:

# Check if executors are underutilized
sc = spark.sparkContext
conf = sc.getConf()

print("=== Resource Allocation ===")
print(f"Driver memory: {conf.get('spark.driver.memory', 'default')}")
print(f"Executor memory: {conf.get('spark.executor.memory', 'default')}")
print(f"Executor cores: {conf.get('spark.executor.cores', 'default')}")
print(f"Dynamic allocation: {conf.get('spark.dynamicAllocation.enabled', 'not set')}")
print(f"Min executors: {conf.get('spark.dynamicAllocation.minExecutors', 'not set')}")
print(f"Max executors: {conf.get('spark.dynamicAllocation.maxExecutors', 'not set')}")

active_execs = len(sc._jsc.sc().getExecutorMemoryStatus()) - 1  # subtract driver
print(f"\nActive executors: {active_execs}")
print(f"Default parallelism: {sc.defaultParallelism}")
print(f"Shuffle partitions: {spark.conf.get('spark.sql.shuffle.partitions', 'default')}")

Pipeline Run Diagnosis

When the user provides a pipeline name or ID with a pipeline run ID (job instance ID), auto-discover all Spark-related activities and diagnose each one.

Trigger examples:

• "Diagnose pipeline ETL_Pipeline run abc-123"
• "My pipeline failed, run ID is abc-123"
• "Check all Spark jobs in pipeline run abc-123"

Step P1 — Resolve Pipeline & Get Run Status

# Resolve pipeline item ID (if user gave name)
pipelineId=$(az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items?type=DataPipeline" \
  --query "value[?displayName=='<PipelineName>'].id" --output tsv)

# Get pipeline run status
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/items/$pipelineId/jobs/instances/$jobInstanceId" \
  --output json

Response includes: id, status (Completed/Failed/InProgress/Cancelled), startTimeUtc, endTimeUtc, failureReason.

Step P2 — Query Activity Runs

Use the public queryActivityRuns API to get all activities within the pipeline run. This returns activity details including Spark session IDs, notebook item IDs, parameters, error details, and tracebacks — everything needed for diagnosis.

API endpoint (docs):

POST $FABRIC_API_URL/workspaces/$workspaceId/datapipelines/pipelineruns/$jobInstanceId/queryactivityruns

Shell escaping: In PowerShell, inline JSON bodies for az rest must use escaped double quotes inside double-quoted strings ("{\"key\":\"value\"}") or use a --body @file.json approach. Single-quoted strings with inner double quotes ('{"key":"value"}') work correctly in PowerShell but may fail if the shell double-processes escapes.

PowerShell `&` in URLs: The & character in URL query parameters (e.g., ?type=driver&fileName=stderr) is interpreted as PowerShell's call operator, breaking the URL. For Spark Monitoring endpoints that require multiple query params, write the URL to a variable first or use --url-parameters:

# BROKEN: PowerShell treats & as call operator
az rest --url "https://...?type=driver&fileName=stderr"

# WORKAROUND: Use a JSON body file or single-param endpoints when possible
# The Spark Monitoring jobs, advice, stagesSummary, and resourceUsage endpoints
# require NO query params — prefer these over log download endpoints.

# Bash — use heredoc or file
cat > /tmp/query-body.json << 'EOF'
{
  "filters": [],
  "orderBy": [{"orderBy": "ActivityRunStart", "order": "DESC"}],
  "lastUpdatedAfter": "<pipeline startTimeUtc minus 1 hour>",
  "lastUpdatedBefore": "<pipeline endTimeUtc plus 1 hour, or now if InProgress>"
}
EOF
az rest --method post --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/datapipelines/pipelineruns/$jobInstanceId/queryactivityruns" \
  --body @/tmp/query-body.json --output json

# PowerShell — escape inner double quotes with backslash
az rest --method post --resource $FABRIC_RESOURCE_SCOPE `
  --url "$FABRIC_API_URL/workspaces/$workspaceId/datapipelines/pipelineruns/$jobInstanceId/queryactivityruns" `
  --body '{\"filters\":[],\"orderBy\":[{\"orderBy\":\"ActivityRunStart\",\"order\":\"DESC\"}],\"lastUpdatedAfter\":\"<startTimeUtc minus 1h>\",\"lastUpdatedBefore\":\"<endTimeUtc plus 1h>\"}' `
  --output json

Response schema — paginated { continuationToken, value: [...] }:

Key insight: The queryActivityRuns response provides both input.notebookId and output.result.sessionId directly — no need to decode the pipeline definition to build an activity→item map. The input.notebookId is the item ID needed for all Spark Monitoring API calls.

Error API priority for pipeline runs — richest to leanest:

Use `queryActivityRuns` first — it has the longest data retention and returns structured error data. Spark Monitoring APIs (Advisor, logs) have shorter TTL and may return 404 for older runs.

Step P3 — Identify Spark Activities & Handle Edge Cases

Filter activities by type to find Notebook and Spark Job Definition activities:

Skip non-Spark activities (e.g., Wait, Copy, SetVariable, ForEach, IfCondition, Until) for diagnostic analysis.

Edge case handling:

Deduplication for ForEach:

# Group activities by activityName
# For each group:
#   If all have same iterationHash → single run (no loop)
#   If different iterationHash values → ForEach loop
#     → Diagnose each iteration
#     → In report, show: "Activity: LoadRawData [iteration 1/5] (Failed)"

Deduplication for retries:

# Group activities by activityName + iterationHash
# For each group:
#   Keep only the run with max(retryAttempt)
#   Note: "Activity retried N times; analyzing final attempt"

Step P4 — Extract Spark Session IDs from Activity Output

For each Notebook/SJD activity found in Step P3, the output and input fields from queryActivityRuns contain everything needed for Spark diagnostics:

Key fields from `queryActivityRuns` response:

Extract livyId and resolve Spark application:

# From the queryActivityRuns response, for each Notebook/SJD activity:
livyId="<activity.output.result.sessionId>"
notebookId="<activity.input.notebookId>"
workspaceId="<activity.input.workspaceId>"  # may differ from pipeline workspace

# Get the Spark application ID for this session
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/notebooks/$notebookId/livySessions/$livyId" \
  --query "sparkApplicationId" --output tsv

No time correlation needed — output.sessionId gives the exact Livy session.

Handling null sessionId (Spark never started):

# When output.result.sessionId is null, Spark never launched.
# Check output.result.error for the Python traceback:
#   output.result.error.ename   — exception class name
#   output.result.error.evalue  — exception message
#   output.result.error.traceback — full stack trace lines
# Also check the pipeline-level error:
#   error.errorCode, error.message, error.failureType
# Common causes:
#   - Library/environment setup failure
#   - Invalid Spark pool configuration
#   - Capacity exhausted (could not allocate executors)
#   - Notebook syntax error caught before submission

Business logic exceptions vs Spark infra failures: When error.failureType is UserError and output.result.error.traceback shows a deliberate raise Exception(...) in user code, this is a business logic failure — Spark ran successfully but the notebook intentionally failed. Report this separately from Spark infrastructure issues. No Spark-level diagnostics are needed for these.

Step P4b — Extract Exact Notebook Cell & Line from Traceback

The output.result.error.traceback array from queryActivityRuns is the primary source for identifying exactly which notebook code failed. Parse the traceback lines to extract the cell number, line number, and failing code.

Traceback format — each entry in traceback[] is a string line. Look for the pattern:

Cell In[<cell_number>], line <line_number>

Example traceback lines:

  File "Cell In[14], line 19"  
    raise Exception(f"Expected key '{key}' not found in URL: {url}")
Exception: Expected key 'cluster_id' not found in URL: https://...

Parsing rules: 1. Cell number — Cell In[14] = notebook cell #14 (1-indexed) 2. Line number — line 19 = line 19 within that cell 3. Exception class — output.result.error.ename (e.g., Exception, ValueError, KeyError) 4. Exception message — output.result.error.evalue — the human-readable error 5. Full stack — walk traceback[] bottom-to-top to find the root cause frame

Construct the Notebook Snapshot URL so users can open the exact failed run in the Fabric UI:

https://app.powerbi.com/workloads/de-ds/sparkmonitor/{notebookId}/{livyId}?trident=1&experience=power-bi&ctid={tenantId}&tab=related

What to report for each failed activity:

#### Activity: <activityName> [iteration <N>/<total>] — FAILED
- **Notebook**: <notebookId> (resolve display name via Items API)
- **Cell**: Cell In[14], line 19
- **Exception**: <ename>: <evalue>
- **Notebook Snapshot**: [Open in Fabric UI](<constructed snapshot URL>)
- **Parameters**: <key pipeline parameters for this iteration>
- **Traceback** (last 5 frames):
  <paste relevant traceback lines>

For ForEach iterations — group by unique error pattern:

• If 31/36 iterations fail with the same ename + same cell/line, report once with count
• Show the unique evalue values (or a sample) to reveal iteration-specific differences
• Example: "31/36 iterations failed at Cell In[14], line 19 with Exception: Expected key 'cluster_id' not found"

Key insight: For user-code failures (business logic, data validation, missing keys), the traceback is the definitive diagnostic — Spark Monitoring APIs will show all jobs SUCCEEDED because the Spark engine ran fine; it was the Python code that raised an exception.

Step P4c — Validate with Spark Monitoring Jobs API

To confirm whether the failure is in Spark infrastructure or user code, check the Spark job-level status:

# Get all Spark jobs for the session
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/notebooks/$notebookId/livySessions/$livyId/applications/$sparkApplicationId/jobs" \
  --output json

Classification logic:

Data TTL: Spark Monitoring data (stages, jobs, executor logs, driver stderr) is retained on the online Spark History Server for a limited time after session completion. For sessions that completed hours or days ago, the monitoring endpoints may return 404. The queryActivityRuns traceback data has a longer retention but also expires (observed: data available for ~1 hour after run, then purged). Always attempt queryActivityRuns first — it is the most reliable data source for recent failures.

Step P5 — Run Standard Diagnostics on Each Session

For each Spark session found (where sessionId is not null and Spark jobs show failures): 1. Apply Step 2 (auto-route by session state) from the main workflow 2. Run Steps 3–6 as applicable (failure analysis, performance, resource, session health) 3. Tag findings with the activity name (+ iteration index if ForEach) for the report

Skip full Spark diagnostics when Step P4c confirms all Spark jobs SUCCEEDED — the traceback from Step P4b is sufficient.

Handle Nested Pipelines (ExecutePipeline activities)

For each ExecutePipeline activity:

# Extract child pipeline run info from activity output
childJobInstanceId="<from activity output>"

# Recurse: run the full Pipeline Run Diagnosis (Steps P1–P6) on the child pipeline
# Use the child pipeline's workspaceId and pipelineId from typeProperties

Report nested pipeline findings under a sub-section: "Child Pipeline: <name> (run <id>)".

Step P6 — Pipeline-Level Report

Present a combined report showing all activities and their Spark diagnostics:

## Pipeline Diagnostic Summary

**Pipeline**: <pipeline name> | **Run ID**: <jobInstanceId> | **Status**: <status>
**Duration**: <startTimeUtc> → <endTimeUtc>

### Activity Overview

| # | Activity | Type | Status | Duration | Spark Issues | Notes |
|---|----------|------|--------|----------|--------------|-------|
| 1 | LoadRawData | TridentNotebook | Failed | 5m 23s | 1 HIGH, 1 MEDIUM | |
| 2 | LoadRawData [iter 2] | TridentNotebook | Succeeded | 4m 10s | 0 HIGH, 1 MEDIUM | ForEach iteration 2 |
| 3 | TransformSilver | TridentNotebook | Succeeded | 12m 45s | 0 HIGH, 2 MEDIUM | Retried 1× |
| 4 | PublishGold | SparkJob | Cancelled | — | — | Upstream failure |
| 5 | RunChildPipeline | ExecutePipeline | Failed | 8m 30s | See child report | Nested pipeline |
| 6 | SetupEnv | TridentNotebook | Failed | 0s | — | Spark never started (library error) |

### Detailed Findings

#### Activity: LoadRawData (Failed)
| # | Severity | Finding | Fix |
|---|----------|---------|-----|
| 1 | HIGH | Executor OOM in stage 4 (skewed partition) | Enable AQE skew join |
| 2 | MEDIUM | Disk spill 1.3 GB in stage 3 | Increase shuffle partitions to 400 |

#### Activity: TransformSilver (Succeeded — performance issues)
| # | Severity | Finding | Fix |
|---|----------|---------|-----|
| 1 | MEDIUM | Data skew ratio 4.1× in stage 7 | Salt the join key |
| 2 | MEDIUM | Core efficiency 28% | Reduce executor count |

#### Activity: SetupEnv (Failed — no Spark session)
**Error**: library packaging error — `pip install failed for package xyz==2.0`
**Fix**: Verify package version compatibility with Fabric runtime Python version.

#### Activity: ValidateData [31/36 iterations FAILED — user-code error]
**Notebook**: Check load_ virtualization (8e28e1fd-...)
**Cell**: Cell In[14], line 19
**Exception**: `Exception: Expected key 'cluster_id' not found in URL`
**Spark Jobs**: 13/13 SUCCEEDED — failure is in Python code, not Spark infrastructure
**Pattern**: 31 iterations fail at the same cell with the same exception class; 5 succeed
**Sample failing parameters**: `cluster_url=https://...`, `batch_key=...`
**Root Cause**: Notebook business logic expects a key in the URL that is missing for certain cluster URLs
**Fix**: Update notebook validation logic to handle URLs without the expected key, or fix upstream data

#### Child Pipeline: IngestPipeline (Failed)
[Recursive report from child pipeline diagnosis]

Prioritization: In the report, order activities by: (1) Failed activities first, (2) Succeeded but slow, (3) Succeeded and healthy. For ForEach iterations, group by activity name and highlight which iterations failed.

Session Health

Scope: Monitor and manage Livy session lifecycle in Microsoft Fabric — detect idle/zombie sessions, assess resource usage, and recover from session failures. All examples use az rest against the Fabric API.

---

Livy Session Lifecycle

Session States

Livy sessions in Fabric follow a defined state machine. Understanding the states and expected transition times is critical for diagnosing stuck sessions.

Expected Startup Times

Starting Stuck Diagnosis

If a session stays in starting beyond the expected time:

# Check how long the session has been starting
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId" \
  --query "{state:state, appId:appId, log:log[-5:]}" --output json

---

Idle and Zombie Session Detection

Sessions that remain idle consume capacity units even when unused. Zombie sessions are sessions that lost their owning process but remain allocated.

List All Sessions with State

Workspace-level listing: To list Spark applications across an entire workspace (not just one lakehouse), use the Workspace and Item-Level Session Listing API which supports filtering by time range, submitter, and application state.

# Get all sessions for a specific lakehouse
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions" \
  --query "sessions[].{id:id, name:name, state:state, appId:appId}" \
  --output table

Identify Idle Sessions

# Filter for idle sessions only
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions" \
  --query "sessions[?state=='idle'].{id:id, name:name, appId:appId}" \
  --output table

Clean Up Idle Sessions

Before cleaning up, verify no active work depends on these sessions:

# Delete a specific idle session
az rest --method delete --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId"

Batch cleanup of all idle sessions:

# List idle session IDs, then delete each
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions" \
  --query "sessions[?state=='idle'].id" --output tsv | while read sid; do
    echo "Deleting idle session: $sid"
    az rest --method delete --resource "$FABRIC_RESOURCE_SCOPE" \
      --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sid"
done

Detect Zombie Sessions

A zombie session is one in idle or busy state but with no recent statement activity. Check the last statement timestamp:

# Get session statements to detect zombies
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/statements" \
  --query "statements[-1:].{id:id, state:state, code:code}" --output json

If the session has zero statements or the last statement completed long ago, the session is likely a zombie — safe to delete.

---

Session Resource Monitoring

Check Session Configuration

# View memory and core allocation for a session
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId" \
  --query "{driverMemory:driverMemory, driverCores:driverCores, executorMemory:executorMemory, executorCores:executorCores, numExecutors:numExecutors}" \
  --output json

Monitor Executor Usage via Livy Statement

Submit a PySpark diagnostic snippet to an idle session:

cat > /tmp/body.json << 'DIAG'
{
  "code": "import json\nsc = spark.sparkContext\nexecs = sc._jsc.sc().getExecutorMemoryStatus()\nresult = {\n    'total_executors': len(execs),\n    'driver_memory_status': str(list(execs.entrySet().iterator().next())),\n    'default_parallelism': sc.defaultParallelism,\n    'dynamic_allocation': sc.getConf().get('spark.dynamicAllocation.enabled', 'unknown')\n}\nprint(json.dumps(result, indent=2))",
  "kind": "pyspark"
}
DIAG
az rest --method post --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/statements" \
  --body @/tmp/body.json --output json

Check Active Capacity Consumption

Session resource usage contributes to capacity consumption. Use the capacity APIs to see overall utilization:

# Get capacity utilization (requires admin permissions)
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/capacities" \
  --query "value[].{name:displayName, id:id, sku:sku, state:state}" \
  --output table

---

Session Recovery Patterns

When to Create a New Session vs Retry

Create Replacement Session

When recreating a session, preserve the original configuration:

# Get the old session's config (if still accessible)
az rest --method get --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$oldSessionId" \
  --query "{driverMemory:driverMemory, driverCores:driverCores, executorMemory:executorMemory, executorCores:executorCores, conf:conf}" \
  --output json > /tmp/old-session-config.json

# Create new session with same config
cat > /tmp/body.json << 'EOF'
{
    "name": "recovery-session",
    "driverMemory": "56g",
    "driverCores": 8,
    "executorMemory": "56g",
    "executorCores": 8,
    "conf": {
        "spark.dynamicAllocation.enabled": "true",
        "spark.fabric.pool.name": "Starter Pool"
    }
}
EOF
az rest --method post --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions" \
  --body @/tmp/body.json --query "{id:id, state:state}" --output json

Session Keep-Alive Pattern

To prevent idle timeout on long-running diagnostic sessions, submit periodic no-op statements:

# Keep-alive: run every few minutes to prevent session timeout
cat > /tmp/body.json << 'EOF'
{"code": "spark.sparkContext.applicationId", "kind": "pyspark"}
EOF
az rest --method post --resource "$FABRIC_RESOURCE_SCOPE" \
  --url "$FABRIC_API_URL/workspaces/$workspaceId/lakehouses/$lakehouseId/$LIVY_API_PATH/sessions/$sessionId/statements" \
  --body @/tmp/body.json --output json

Spark History Server — Local Setup Reference

Scope: Start a local OSS Spark History Server to view the full Spark UI (DAG, tasks, SQL plans) for Fabric Spark applications whose event logs have been downloaded to local disk. This is the Tier 2 offline fallback when online Monitoring APIs are insufficient.

---

Overview

The Spark History Server (SHS) is a standalone web UI for viewing completed Spark application event logs. It reconstructs the Spark UI from event log files, allowing post-mortem analysis of jobs, stages, tasks, SQL queries, and executor metrics.

When to use: Escalate from Tier 1 (online APIs) when you need the full DAG visualization, task-level detail, or SQL plan visualizations that the REST API cannot provide. See Diagnostic Tiers.

---

Prerequisites

---

Key Configuration Properties

---

Starting the History Server

Unix/Linux/macOS

$SPARK_HOME/sbin/start-history-server.sh

Windows (via spark-class)

%SPARK_HOME%\bin\spark-class.cmd org.apache.spark.deploy.history.HistoryServer

With Custom Config

Create a spark-defaults.conf with your settings and point to it:

export SPARK_CONF_DIR=/path/to/conf
$SPARK_HOME/sbin/start-history-server.sh

Example spark-defaults.conf:

spark.history.fs.logDirectory=file:///home/user/.spark-local/event-logs
spark.history.ui.port=18080

---

Stopping the History Server

Unix/Linux/macOS

$SPARK_HOME/sbin/stop-history-server.sh

Windows

Kill the Java process by PID, or use Task Manager.

---

Event Log Directory Format

The SHS expects event log files in the configured logDirectory. Each Spark application writes one event log file (or directory):

spark-events/
  local-1234567890123/     # directory-based log
    events_1_...           # event data
  application_1234567890123_0001   # single-file log

Using Local File Paths on Windows

When using file:/// URIs on Windows, convert backslashes to forward slashes:

# Correct
spark.history.fs.logDirectory=file:///C:/Users/me/.spark-local/event-logs/abc123

# Wrong — backslashes will cause errors
spark.history.fs.logDirectory=file:///C:\Users\me\.spark-local\event-logs\abc123

---

Workflow: From Fabric Event Logs to Local SHS

Step 1 — Copy Event Logs from Fabric

Use the JobInsight API to copy event logs to a OneLake lakehouse:

import com.microsoft.jobinsight.diagnostic.LogUtils
LogUtils.copyEventLog(workspaceId, artifactId, livyId, "sessions", targetDir, true, 1)

Step 2 — Download from OneLake

Download the event log files from OneLake DFS to local disk:

# Get token for OneLake
TOKEN=$(az account get-access-token --resource https://storage.azure.com --query accessToken -o tsv)

# List files in event log directory
curl -H "Authorization: Bearer $TOKEN" \
  "https://onelake.dfs.fabric.microsoft.com/$workspaceId/$lakehouseId/Files/spark-events/$livyId?resource=filesystem&recursive=true" | jq '.paths[].name'

# Download each file
curl -H "Authorization: Bearer $TOKEN" \
  "https://onelake.dfs.fabric.microsoft.com/$workspaceId/$lakehouseId/Files/spark-events/$livyId/<filename>" \
  -o ~/.spark-local/event-logs/<filename>

Step 3 — Start Local SHS

# Point SHS to the downloaded event logs
export SPARK_HOME=~/.spark-local/spark-4.1.1-bin-hadoop3
export SPARK_CONF_DIR=~/.spark-local/conf

cat > $SPARK_CONF_DIR/spark-defaults.conf << EOF
spark.history.fs.logDirectory=file:///$HOME/.spark-local/event-logs
spark.history.ui.port=18080
EOF

$SPARK_HOME/sbin/start-history-server.sh

Step 4 — Open Spark UI

Navigate to http://localhost:18080. The application should appear in the list. Click through to:

• Jobs tab — overview of all Spark jobs
• Stages tab — detailed stage metrics, task distribution
• Executors tab — memory/disk/GC per executor
• SQL tab — SQL plan visualization with metrics
• Environment tab — Spark configuration snapshot

---

Common Issues

"No completed applications found"

• The event log directory is empty or contains no valid event logs
• Check that spark.history.fs.logDirectory points to the correct path
• Ensure the files are Spark event logs (not arbitrary JSON/text files)

"Port already in use"

• Another process is using port 18080
• Change the port: spark.history.ui.port=18081
• Or stop the existing process

Java Not Found

• SHS requires Java 11+ (Java 17 recommended for Spark 4.x)
• Set JAVA_HOME or ensure java is on PATH

OutOfMemoryError on Large Logs

• Very large event logs (>1 GB) may require more heap memory
• Set SPARK_DAEMON_MEMORY before starting:

  export SPARK_DAEMON_MEMORY=4g

Windows-Specific Issues

• Long paths: Event log paths exceeding 260 characters may fail. Use short directory names or enable Windows long path support.
• File locking: If files are locked by another process, SHS cannot read them. Ensure no other application has the event logs open.
• Firewall: Windows Firewall may block the SHS port. Allow java.exe through the firewall or add a port exception.

---

Spark 4.x Changes

Spark 4.x (including 4.1.1 used by Fabric) introduces:

• Improved Structured Streaming UI in History Server
• Better support for Spark Connect session history
• Enhanced SQL/DataFrame metrics visualization
• Requires Java 17 (Java 11 minimum)