Maintaining Fabric Pipelines

Python
# Check pipeline run status and errors
from notebookutils import mssparkutils

# Get recent pipeline runs
runs = mssparkutils.fabric.list_runs(pipeline_name="your_pipeline")
failed_runs = [r for r in runs if r['status'] == 'Failed']

# Get detailed error info
for run in failed_runs[:5]:  # Last 5 failures
    error_details = mssparkutils.fabric.get_run_details(run['runId'])
    print(f"Run {run['runId']}: {error_details['error']}")

Progress:

• Monitor Pipeline Health - Check run history, success rates, duration trends
• Identify Issues - Analyze failed runs, performance bottlenecks, data quality problems
• Diagnose Root Cause - Review activity logs, data lineage, resource usage
• Implement Fixes - Update pipeline logic, adjust configurations, optimize queries
• Test & Deploy - Validate fixes in dev environment, deploy to production
• Document Changes - Update runbooks, create alerts for similar issues

Issue Diagnosis Steps

1. Check Activity Logs

   • Navigate to Monitor > Pipeline runs
   • Click failed run → View details → Activity logs
   • Look for error patterns in timestamps

2. Analyze Data Flow

   • Verify source data availability and format
   • Check transformation logic for edge cases
   • Validate destination connectivity and permissions

3. Review Resource Usage

   • Check Spark cluster scaling and memory usage
   • Monitor concurrent pipeline execution limits
   • Verify compute capacity during peak hours

Example 1: Memory Error Fix Input: Pipeline failing with "OutOfMemoryError" in data transformation Output:

JSON
// Update pipeline activity settings
{
  "typeProperties": {
    "sparkConfig": {
      "spark.executor.memory": "8g",
      "spark.executor.cores": "4",
      "spark.dynamicAllocation.maxExecutors": "20"
    }
  }
}

Example 2: Data Quality Issue Input: Pipeline completing but producing duplicate records Output:

SQL
-- Add deduplication step in data transformation
SELECT DISTINCT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY key_column ORDER BY updated_date DESC) as rn
  FROM source_table
) ranked
WHERE rn = 1