One pipeline, many runs

Copying a pipeline to change one filter is how a repository ends up with fourteen near-identical files that drift apart. Parameters exist to prevent exactly that.

The pattern

{
  "name": "regional_orders",
  "input": { "format": "delta", "path": "sales.orders" },
  "transformations": [
    { "type": "filter", "condition": "region = '{region}' AND ordered_at >= '{since}'" },
    { "type": "filter", "skip_if_false": "{products}", "condition": "product_id IN ({products})" },
    { "type": "stop_if_empty", "message": "Nothing for {region} since {since}" }
  ],
  "output": { "format": "delta", "path": "analytics.orders_{region}", "mode": "overwrite" }
}

for region in ["br", "pt", "es"]:
    fw.run("regional_orders.json", params={
        "region": region,
        "since": since,
        "products": [],        # empty → the product filter is skipped entirely
    })

One file, one code path, three jobs. Fixing a bug fixes it everywhere.

What each type becomes

Python	In the file	Use
`"br"`	`br`	a path fragment, a literal in SQL
`42`	`42`	a threshold
`True`	`true`	keeps a step guarded by `skip_if_false`
`False`	(empty)	skips that step
`["A", "B"]`	`'A', 'B'`	a SQL `IN (...)`
`[1, 2]`	`1, 2`	a numeric `IN (...)`
`[]`	(empty)	falsy — skips the step

Switching steps on and off

skip_if_false reads the value after substitution:

// on/off from a boolean parameter
{ "type": "join", "skip_if_false": "{enrich}", "with": { … }, "on": "customer_id" }

// only when a value was provided
{ "type": "filter", "skip_if_false": "{region}", "condition": "region = '{region}'" }

// branch on which flow is running
{ "type": "struct", "skip_if_false": "'{flow}' in ('ISSUE', 'ISSUE_AND_REGISTER')", "column": "payload", "fields": { … } }

That third form is what replaces a pile of near-duplicate files: one pipeline covering several flows, with the differences expressed as conditions instead of copies.

{
  "transformations": [
    { "$include": "shared/standard_filters.json" },
    { "type": "with_column", "column": "revenue", "expression": "quantity * unit_price" }
  ]
}

[
  { "type": "filter", "condition": "status = '{status}'" },
  { "type": "drop_duplicates", "columns": ["id"] }
]

Fragments are parameterized like anything else — substitution happens after the include is expanded. The path is relative to the pipeline file, includes do not nest, and the directive works only in the top-level transformations.

Where the values come from

Keep them outside the pipeline and outside your code:

import os
from datetime import date

fw.run("regional_orders.json", params={
    "region": os.environ["REGION"],
    "since": date.today().isoformat(),
    "enrich": os.environ.get("ENRICH", "false") == "true",
})

On Databricks, job parameters and widgets map naturally onto this dictionary. In Airflow, they come from the DAG run configuration. That is what makes one file serve a scheduled load, a backfill and a debug run.

Declare them in Studio

Studio lists a Job’s parameters and renders an input for each — typed as text, number, switch or chips — and sends them with a local run. It also lints the two mismatches that bite: a {param} used but never declared, and a declared parameter nothing uses.

Keep the file honest

Name for meaning, not position: {region}, not {p1}.
Document them in the pipeline description — the file is the documentation.
Do not parameterize the shape. Parameters are values. When two variants need different structures, skip_if_false is the tool; when they need different pipelines, write two.
Never put a secret in a parameter that ends up committed. Read credentials from the environment in the job that launches the run, and pass only non-sensitive values here.