Transformations

Transformations are the entries of the transformations array. They run in the order you write them, each receiving the DataFrame the previous one returned.

Every one accepts the meta key skip_if_false, which turns the step on or off per run.

Filtering and shaping

filter

Keeps the rows matching a SQL boolean expression.

{ "type": "filter", "condition": "status = 'CONFIRMED' AND amount > 0" }

Key	Type	Required
condition	SQL expression	yes

Common carrier of {{runtime}} placeholders: "id IN ({{ids}})".

select

Projects a list of columns — or full SQL expressions with aliases.

{ "type": "select", "columns": ["id", "customer", "to_json(payload) AS value"] }

Caution

Every entry goes through F.expr, not F.col. A column whose name has spaces, dots or reserved words must be backticked: "`my column`". A dotted name like a.b is read as a struct field, not as a column literally named a.b.

drop

Removes columns. Names that do not exist are ignored silently by Spark.

{ "type": "drop", "columns": ["tmp_flag", "debug_note"] }

rename

Renames columns through an ordered map, applied one at a time.

{ "type": "rename", "mappings": { "created_at": "creation_date", "nm": "name" } }

Because it is sequential, chained renames work (a→b then b→c ends as c) and reordering the keys changes the result.

cast

Casts columns to Spark types.

{ "type": "cast", "columns": { "amount": "decimal(18,2)", "ordered_at": "date" } }

Uses F.col, so the column must exist. Values that cannot be cast become null — Spark does not raise.

sort

Orders the DataFrame.

{ "type": "sort", "columns": ["ordered_at", "id"], "ascending": true }

ascending accepts a boolean or a list of booleans, one per column.

drop_duplicates

Removes duplicate rows, optionally scoped to a subset.

{ "type": "drop_duplicates", "columns": ["id"] }

columns is optional: omitting it (or passing an empty list) deduplicates over all columns, exactly like distinct. Which row survives is not deterministic — sort first if it matters.

distinct

Removes duplicate rows using every column. No parameters.

{ "type": "distinct" }

fill_na

Replaces nulls.

{ "type": "fill_na", "value": 0, "columns": ["quantity"] }
{ "type": "fill_na", "value": { "quantity": 0, "segment": "UNKNOWN" } }

The scalar form takes an optional columns subset. Spark ignores fills whose type does not match the column type — a silent no-op, not an error.

Computing columns

with_column

Adds or replaces computed columns from SQL expressions. Two mutually exclusive forms.

// single column
{ "type": "with_column", "column": "revenue", "expression": "quantity * unit_price" }

// several, in order — later expressions can use earlier ones
{ "type": "with_column", "columns": {
    "revenue": "quantity * unit_price",
    "revenue_brl": "revenue * fx_rate"
  } }

Precedence

Whenever the columns key is present, it wins and column / expression are ignored. Never emit both. Key order is load-bearing: columns are created in map order, which is why revenue_brl above can reference revenue.

name is accepted as a legacy alias of column, parsed for backward compatibility; write column in new pipelines.

struct

Builds a nested struct column from a field map. More readable than a hand-written named_struct.

{
  "type": "struct",
  "column": "payload",
  "fields": {
    "external_id": "contract_id",
    "issuer.name": "issuer_name",
    "issuer.document": "lpad(cast(document as string), 14, '0')",
    "amounts": { "principal": "principal_amount", "interest": "interest_amount" }
  }
}

• String values are SQL expressions; object values nest further.
• Dot-paths auto-nest: issuer.name and issuer.document become one issuer struct, so the payload reads like a flat table in the file and arrives nested in the data.
• Field order follows key order, which keeps diffs stable.
• A field name containing a literal dot is impossible — every dot means nesting.

Two conflicts raise at apply time: using a path segment that is already a leaf (a used as both value and prefix), and duplicating a leaf key.

sql

Runs arbitrary Spark SQL over the current DataFrame.

{
  "type": "sql",
  "view_name": "_df",
  "query": "SELECT customer, SUM(revenue) AS total FROM _df GROUP BY customer"
}

The DataFrame is registered as a temp view named view_name (default _df) and the query result becomes the new DataFrame. The escape hatch for anything the other transformations do not express.

Aggregating

group_by

Groups and aggregates, with optional pivot.

{
  "type": "group_by",
  "by": ["customer_id", "country"],
  "agg": [
    "sum(revenue) as revenue_total",
    "count(*) as orders",
    "max(ordered_at) as last_order"
  ],
  "pivot": { "column": "month", "values": ["jan", "feb", "mar"] }
}

Key	Type	Required
by	list of columns	yes
agg	list of complete SQL aggregate expressions, with aliases	yes
pivot	column name, or { column, values }	no

Caution

agg is a list of strings, never a map. Listing values on a pivot saves Spark an extra scan to discover them; an empty list behaves as if the key were absent. Every column not named in by or produced by agg disappears.

Combining sources

join

Joins a second source read inline.

{
  "type": "join",
  "with": { "format": "delta", "path": "sales.customers" },
  "with_transformations": [
    { "type": "filter", "condition": "active = true" },
    { "type": "select", "columns": ["customer_id", "segment"] }
  ],
  "on": "customer_id",
  "how": "left"
}

Key	Type	Notes
with	source config	any readable format
on	string, list, or SQL expression	"id", ["a","b"], or "l.id = r.id AND l.dt = r.dt"
how	join type	inner (default), left, right, full, cross, leftsemi, leftanti, …
broadcast	true / "right" / "left" / false	map-side (broadcast) join hint
with_transformations	list	applied to the right side before the join

The left DataFrame is aliased l and the right one r, so an expression on can disambiguate columns. Inside with_transformations the aliases do not exist yet — use bare column names there.

Broadcast join

Set broadcast when one side is small enough to fit in each executor’s memory (a dimension, a lookup). Spark ships that side to every executor and joins map-side, skipping the shuffle of the large side entirely.

{
  "type": "join",
  "with": { "format": "delta", "path": "ref.dim_product" },
  "on": "product_id",
  "how": "left",
  "broadcast": true
}

Value	Broadcasts
true or "right"	the second source (the with side — the small dimension/lookup)
"left"	the main DataFrame
false or absent	no hint — Spark decides by size

Broadcasting a side that is not actually small can exhaust executor memory; leave it absent when in doubt.

In Studio, with and with_transformations come from the node’s second input, not from a form field.

union

Appends the rows of another source.

{
  "type": "union",
  "with": { "format": "parquet", "path": "/data/orders_archive" },
  "allow_missing_columns": false
}

Positional by default

With allow_missing_columns: false (the default) Spark matches columns by position, not by name — a classic silent-corruption source when the two schemas drifted. Set it to true to match by name and fill absent columns with null.

union has no with_transformations: the right side is read as-is.

Layout and file count

repartition

Redistributes the DataFrame’s partitions — it changes cost, never data.

{ "type": "repartition", "num_partitions": 64, "columns": ["pmod(hash(id), 64)"] }

Key	Values	Default
num_partitions	target number of partitions (positive integer)	—
columns	column names or SQL expressions; equal values land in the same partition	—
coalesce	merge without a shuffle — reduces only	false
range	repartitionByRange: split by value band instead of by hash	false

At least one of num_partitions and columns is required.

This is the missing piece between partition_by on the output, which decides which directories exist, and the number of files written, which nobody was deciding. A file is written per (task, directory) pair that holds rows, so 200 shuffle partitions over 30 days of dt leave up to 6,000 files behind. Repartitioning by the same expressions the output partitions by collapses that to one file per key value, because a single key value never splits across tasks — AQE merges neighbouring partitions but never separates one.

[
  { "type": "repartition", "columns": ["dt"] },
  { "type": "repartition", "num_partitions": 1, "coalesce": true },
  { "type": "repartition", "num_partitions": 8, "columns": ["event_date"], "range": true }
]

Every invalid combination raises with the reason instead of quietly doing something else: coalesce with columns (no key to group by), coalesce with range, coalesce without a count, range without columns, a non-integer or non-positive count, and no parameters at all.

Control and inspection

checkpoint

Materializes the DataFrame and truncates its logical plan.

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

Key	Values	Default
method	localCheckpoint, checkpoint	localCheckpoint
eager	boolean	true

Use it after heavy joins, before a collect, and before fanning out to several destinations — it stops Spark from recomputing the same lineage repeatedly. An invalid method is ignored, with a warning emitted at the end of the run.

collect

Collects a column’s distinct values into a runtime variable.

{ "type": "collect", "column": "customer_id", "as": "active_customers" }

The DataFrame passes through unchanged, but the values land in {{active_customers}} for later steps. It triggers a driver-side action, so run it after a checkpoint. See runtime variables.

Key	Values	Default
column	the column whose distinct values are collected	—
as	runtime variable name	—
max_values	ceiling on the number of distinct values; 0 disables it	10000

The collected list becomes a literal inside IN (...), and past a few thousand values the remedy becomes the problem: the plan grows, Catalyst spends its time analysing the predicate and the pushdown degrades. Over max_values the step fails and names the alternative — a semi/inner join against the list as a DataFrame, which Spark resolves as a broadcast join without the driver ever holding the values. The ceiling is applied inside the query (limit(max_values + 1)), so a column with millions of distinct values is never materialized in the driver, and above 1,000 values it still works but warns.

stop_if_empty

Ends the run gracefully when there is nothing to process.

{ "type": "stop_if_empty", "message": "No approved orders in the window" }

The remaining transformations and every write are skipped. The result comes back with skipped: true, success: true and rows_written: 0 — a no-op, not a failure. Place it right after the filter that defines the working set, before expensive joins.

Caution

Avoid it inside a per-destination chain: the run is reported as skipped even though earlier destinations were already written.

debug

Inspects the DataFrame without changing it.

{
  "type": "debug",
  "label": "after enrichment",
  "actions": ["count", "print_schema", "show"],
  "transformations": [{ "type": "filter", "condition": "id = 'X1'" }],
  "show_rows": 20,
  "truncate": true,
  "vertical": false,
  "extended": false
}

actions accepts count, print_schema, show, explain, pushdown, columns, dtypes. Its nested transformations apply to a throwaway copy used only for the inspection — the pipeline DataFrame always passes through untouched.

pushdown

Answers the question explain only hints at: what actually reached the source?

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

It reads the physical plan and reports, per read node, PartitionFilters (partitions pruned before a file is opened), PushedFilters (the predicate handed to Parquet/ORC or to the database), PushedAggregates, PushedGroupBy, RuntimeFilters (dynamic partition pruning and join bloom filters) and how many columns the scan returns. A scan that pushed nothing is called out with what to do about it, and Filter nodes sitting above the scans are counted — a predicate evaluated after the read is data that came off disk to be thrown away. It fires no job: the physical plan is planning, not execution.

$include

Not a transformation but a directive: it inlines a JSON fragment.

{ "$include": "shared/standard_filters.json" }

The path is relative to the pipeline file. The fragment may be a single object or a list. Nested includes are not expanded, and the directive works only in the top-level transformations array.

Quick index

Type	Purpose
filter	keep matching rows
select	project columns or expressions
drop	remove columns
rename	rename columns
cast	change column types
with_column	compute columns
struct	build a nested column
drop_duplicates	deduplicate, optionally by subset
distinct	deduplicate over all columns
sort	order rows
fill_na	replace nulls
sql	arbitrary Spark SQL
group_by	aggregate, with optional pivot
join	join a second source
union	append another source
repartition	redistribute partitions, control the number of files
checkpoint	materialize and truncate the plan
collect	publish a runtime variable
stop_if_empty	end the run when there is no data
debug	inspect without changing