Outputs

A pipeline writes one destination with output, or several with outputs. Each destination is configured independently, and they all start from the same transformed DataFrame.

{
  "outputs": [
    { "format": "delta", "path": "analytics.orders", "mode": "overwrite" },
    { "format": "csv", "path": "/exports/summary", "columns": ["id", "revenue"] }
  ]
}

Fields

Key	Type	Default	Purpose
format	string	—	connector, lowercased at parse
path	string	—	path, table, view or topic
mode	string	overwrite	write behavior
partition_by	list	[]	physical partitioning, file formats only
columns	list	all	column projection for this destination
options	object	{}	connector options
transformations	list	[]	reshaping for this destination only

Write modes

Mode	Behavior
overwrite	replaces the destination
append	adds rows
merge	upsert — Delta and Iceberg

merge requires options.merge_keys:

{
  "format": "delta",
  "path": "analytics.orders",
  "mode": "merge",
  "options": {
    "merge_keys": ["order_id"],
    "merge_condition": "T.deleted = false"
  }
}

T is the target table, S the incoming DataFrame.

Caution

Delta raises multiple source rows matched when the incoming DataFrame has more than one row per merge key. Add a unique validation on those columns so the pipeline tells you before the write does.

Column projection

columns selects what this destination writes, without touching the others:

{
  "outputs": [
    { "format": "parquet", "path": "/dw/orders_full" },
    { "format": "csv", "path": "/exports/finance", "columns": ["order_id", "revenue", "country"] }
  ]
}

Note

These are plain column names, not SQL expressions — the projection is a df.select(*columns). A name that does not exist raises and fails the whole pipeline. For computed columns, use transformations below.

Per-destination transformations

When destinations need different shapes — not just different columns — give each one its own chain. It runs after the validations, on that destination only.

{
  "outputs": [
    {
      "format": "kafka",
      "path": "orders-events",
      "transformations": [
        { "type": "struct", "column": "payload", "fields": { "id": "order_id", "total": "revenue" } },
        { "type": "with_column", "column": "value", "expression": "to_json(payload)" }
      ],
      "options": { "bootstrap_servers": "broker:9092", "value_column": "value" }
    },
    {
      "format": "delta",
      "path": "analytics.customer_revenue",
      "mode": "merge",
      "transformations": [
        {
          "type": "group_by",
          "by": ["customer_id"],
          "agg": ["sum(revenue) as revenue", "count(*) as orders"]
        }
      ],
      "options": { "merge_keys": ["customer_id"] }
    }
  ]
}

Every transformation type is available here, including join and runtime {{variables}}.

Checkpoint before fanning out

Each destination re-reads the same DataFrame. Without a checkpoint as the last main transformation, Spark recomputes the whole lineage once per destination — the single most common cause of a pipeline that is N times slower than it should be.

Order of operations per destination

For each destination, in this order:

1. its own transformations
2. the columns projection
3. the write

Which means a projection sees the columns the destination’s own transformations produced, not the ones the main chain ended with.

Staging between pipelines

Temp views turn a set of pipelines into a composable job: several pipelines write the same staging view, and a final one validates and publishes it.

// pipeline 1..N
{ "output": { "format": "view", "path": "orders_staging", "mode": "overwrite" } }

// final pipeline
{ "input": { "format": "view", "path": "orders_staging" } }

All of them must run in the same Spark session — the same process — which is exactly what Sparquet does when you call run() several times on one instance.

What comes back

result = fw.run("pipeline.json")
result.rows_written        # total across every destination

for m in result.output_metrics:
    print(m.format, m.mode, m.path, m.rows_written)

Each destination reports its own OutputMetrics (format, path, mode, rows_written) in result.output_metrics. The count is taken on that destination’s final DataFrame — after its own transformations and column projection, just before the write — so it is exact even when a per-destination chain explodes or aggregates rows.

Note

result.rows_written is the sum of those per-destination counts, not a single measurement of the main DataFrame. A destination that aggregates rows contributes its own reduced count to the total.