Spark CLI

# 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')}")

# 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)")

# 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")

# 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')")

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

# 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/")

# 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")

# 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

# 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/")

# 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()

# 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())}")

# 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)

# 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

# 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)

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

# 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()

# 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')}")

# 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

# 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

# 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

# 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')}")