Performance

Most slow pipelines are slow for one of four reasons: they read too much, they recompute the same work, they move data to the driver, or they fan out without materializing. Each has a direct fix in the language.

Read less

The cheapest row is the one that never leaves the source.

// partitioned lake table: filter on the partition column first
{ "type": "filter", "condition": "ordered_at >= '{since}'" }

// then narrow the columns, before any join
{ "type": "select", "columns": ["id", "customer_id", "amount"] }

Order matters: a select after the join still carries every column through the shuffle.

Make it a habit: start every chain with filter then select, before any join, struct, group_by or window. Spark’s optimizer pushes some of this down on its own, but doing it explicitly shrinks the data for every step that follows, helps the planner, and makes the pipeline read as “narrow early, work later.”

Reuse the input without re-reading

When you need to self-join the input — or run SQL against it — without reading the source twice, register it once as a cached temp view with input_view:

fw.run("orders.json", input_view="orders")            # session temp view, cached
fw.run("orders.json", input_view={"name": "orders", "type": "global"})  # global_temp.orders

// a later step joins the input to itself through the view — no second read
{ "type": "join",
  "with": { "format": "view", "path": "orders" },
  "with_transformations": [
    { "type": "group_by", "by": ["customer_id"], "agg": ["max(amount) as top"] }
  ],
  "on": "customer_id", "how": "left" }

The input DataFrame is cache()d and registered before the transformations run, so the self-join reuses the cached rows instead of recomputing the source lineage. Set it in the Sparquet(...) constructor to apply to every run, or per run(...) call.

Checkpoint the plan

{ "type": "checkpoint", "method": "localCheckpoint", "eager": true }

Three places earn it:

• After heavy joins, so the plan does not keep growing.
• Before collect, which is a driver action that would otherwise recompute the whole lineage.
• Before fanning out to several destinations, so the work happens once instead of once per destination.

The single biggest win

A pipeline with three outputs and no checkpoint computes its entire lineage three times. Adding one node usually cuts the runtime by two thirds.

Push the key list down

When the working set is small and the table you are enriching from is enormous:

[
  { "type": "checkpoint" },
  { "type": "collect", "column": "customer_id", "as": "customers" },
  {
    "type": "join",
    "with": { "format": "delta", "path": "sales.bronze_events" },
    "with_transformations": [
      { "type": "filter", "condition": "customer_id IN ({{customers}})" }
    ],
    "on": "customer_id"
  }
]

The literal IN (...) lets Delta and Parquet skip files using their statistics. Keep the list to tens of thousands of keys at most — beyond that, a plain join is the right tool. See Joins and pushdown.

Prove the pushdown

{ "type": "debug", "label": "read", "actions": ["pushdown"] }

explain hints at it; this answers it. Per read node it prints PartitionFilters, PushedFilters, PushedAggregates, PushedGroupBy, RuntimeFilters and how many columns the scan returns, and it counts the Filter nodes sitting above the scans — a predicate evaluated after the read is data that came off disk to be thrown away. What it makes visible:

• a filter on a partition column lands in PartitionFilters and skips whole directories; a filter on a regular column lands in PushedFilters and only skips row groups whose min/max statistics rule them out.
• a predicate the source cannot express — a UDF, a cast, lower(col) = 'x' — stays as a Filter above the scan. Rewrite it so the column is bare on one side.
• for JDBC, only table reads push. With query the SELECT you wrote is already the cut.
• RuntimeFilters is dynamic partition pruning: the fact table pruned by the dimension’s filter, decided at runtime.

It costs nothing — reading the physical plan is planning, not execution.

File-level levers

The defaults are right for most pipelines. These are the ones worth knowing when they are not; set them under spark.configs.

Config	Default	What it does
spark.sql.parquet.filterPushdown	true	pushes the predicate into Parquet’s row-group statistics
spark.sql.parquet.aggregatePushdown	false	answers min/max/count from the footer, without reading data
spark.sql.orc.filterPushdown / .aggregatePushdown	true / false	the same, for ORC
spark.sql.optimizer.dynamicPartitionPruning.enabled	true	prunes the fact table with the dimension’s filter at runtime
spark.sql.optimizer.runtime.bloomFilter.enabled	true	bloom filter on the join key, cuts what gets shuffled
spark.sql.files.maxPartitionBytes	128MB	task size when reading files
spark.sql.adaptive.coalescePartitions.enabled	true	merges tiny post-shuffle partitions

aggregatePushdown is the one worth trying by hand: with it on, a count(1) or a max(dt) over Parquet is answered from the file footers and reads no data at all — but it is off by default because it only applies to a bare aggregation over the scan, and any filter or cast in between silently gives the gain back.

Reader options in the same family: mergeSchema (off by default, and each extra file read costs a listing), basePath (which prefix the partition columns are derived from), recursiveFileLookup and pathGlobFilter (read less by not listing it in the first place), and spark.sql.sources.partitionOverwriteMode: dynamic, which makes overwrite replace only the partitions present in the DataFrame instead of the whole directory.

Stop early

{ "type": "stop_if_empty", "message": "Nothing to process" }

Placed right after the filter that defines the working set, it turns a no-op night into seconds instead of a full pipeline over zero rows.

Shape the right side of a join

"with_transformations": [
  { "type": "filter", "condition": "active = true" },
  { "type": "select", "columns": ["customer_id", "segment"] },
  { "type": "distinct" }
]

Filter, project, deduplicate — in that order — before the join. A right side that repeats the key multiplies rows, which is both a correctness bug and a performance one.

Skew is two different problems

• Read-side join skew — one key with far more rows than the rest leaves a single task running for minutes while the others are done. AQE handles this one: spark.sql.adaptive.skewJoin.enabled (default true), with skewedPartitionFactor (5) and skewedPartitionThresholdInBytes (256MB) deciding what counts as skewed; it splits the offending partition and replicates the matching side. It only fires on a sort-merge join with AQE on, so a join that was hinted to broadcast never reaches it.
• Write-side skew — one value of a partition_by column holding most of the rows writes one enormous file while the rest of the directories are finished. AQE cannot help here, because the directory layout is the requirement, not an accident of the plan. Split it deliberately: repartition by the partition columns plus a salt (pmod(hash(id), 8)), which turns that directory’s single file into eight, and remember that the reader pays nothing for the extra files as long as they stay large.

Watch the write

• partition_by with high cardinality creates thousands of tiny files. Partition on something coarse (date, country), never on an id.
• merge costs more than append. Use it when you need idempotence, not by default.
• CSV inferSchema costs an extra pass. Set explicit cast transformations and turn it off for large, stable feeds.

DataFusion Comet, opt-in

Comet is a Spark plugin that replaces physical operators with native (Rust/Arrow) implementations and falls back to Spark for what it does not support. The framework needs no code for it — it is session config:

Why turn it on. The benefit is not “Rust is fast” — it is four concrete properties:

• Vectorized execution on Arrow. Native operators work on columnar batches with SIMD instructions instead of row by row on the JVM. That is why the gain shows up in scan, filter, projection, hash aggregate and sort, and disappears where the time is spent on the network.
• Off-heap memory. Execution buffers leave the JVM heap, which takes the heaviest allocator away from the garbage collector. On a job with visible GC pauses in the Spark timeline, this is the half of the gain you notice first.
• No change to the pipeline. It is session configuration, not API: same JSON, same readers, writers, transformations and validations. Nothing in the pipeline file says Comet exists.
• Partial and reversible adoption. The swap is per operator, and what Comet does not support stays on Spark inside the same plan. There is no migration and no one-way door: remove the configs and the next run is back to where it was.

It pays off on heavy Parquet reads with filters, projections, aggregations, sorts or shuffles in the way — the shape of most batch ingestion. It does not pay off when the time is on the other side of the network (JDBC, APIs), on Python UDFs (which force that stretch back to Spark), on write-dominated pipelines, or on small volumes, where sizing off-heap and loading an 88 MB jar costs more than it saves.

{
  "spark": {
    "configs": {
      "spark.plugins": "org.apache.spark.CometPlugin",
      "spark.shuffle.manager": "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
      "spark.memory.offHeap.enabled": "true",
      "spark.memory.offHeap.size": "4g"
    }
  }
}

Measured on this framework’s own pipeline, 40,000,000 rows / 290 MB of Parquet, local[4], three timed repetitions after a discarded warm-up (median):

Shape	Without Comet	With Comet	Gain
filter + group_by with sum/count	3.56s	1.61s	2.21x
filter + count	1.02s	0.83s	1.24x

The gain tracks how much of the plan went native: the aggregation came back with CometNativeScan, CometFilter, CometHashAggregate, CometExchange and CometNativeShuffle, while the count shape has far less to accelerate. Before turning it on:

• the jar has to be on the driver classpath before the JVM starts — --driver-class-path, spark.driver.extraClassPath, or the cluster’s jar directory. Through spark.jars it arrives too late and the session dies with java.lang.ClassNotFoundException: org.apache.spark.CometPlugin.
• the artifact is per Spark line: org.apache.datafusion:comet-spark-spark4.1_2.13:1.0.0 for Spark 4.1, about 88 MB.
• the native library ships for Linux only. Anywhere else the plugin loads, disables itself silently and the pipeline passes with no acceleration at all — so check the plan for Comet* nodes instead of assuming.
• off-heap memory is mandatory (spark.memory.offHeap.enabled and .size).
• fallback is per operator, and spark.comet.explain.fallback.enabled logs the reason for each one.

Measure before guessing

{ "type": "debug", "label": "after join", "actions": ["count", "explain"], "extended": false }

explain shows the plan Spark will actually run — the fastest way to confirm a filter was pushed down or a broadcast happened. count after each stage tells you where rows multiplied.

The Run panel in Studio reports rows read, rows written and duration per run, which is often enough to spot the regression without opening the Spark UI.

A quick checklist

Symptom	Likely cause
Runtime scales with the number of outputs	Missing checkpoint before the fan-out
Driver out of memory	collect on a large column, or a huge IN (...) list
Join takes longer than the whole rest	Right side not filtered or projected before the join
Thousands of tiny files	partition_by on a high-cardinality column
Slow every night, even with no data	Missing stop_if_empty
A Filter sits above the scan in the plan	Predicate the source cannot express — check with debug / pushdown
One task runs for minutes while the rest are done	Join skew (AQE), or one partition value holding most of the rows (write side)