Why did my SDP flow fail?

Summary

• In Spark Declarative Pipelines (SDP), a flow reads from a source, applies your logic and writes to a target dataset. The Apache Spark guide documents the rules to check first.
• Run spark-pipelines dry-run to catch syntax errors, analysis errors and cyclic dependencies without reading or writing data.
• The guide says collect(), count(), pivot(), toPandas(), save(), saveAsTable(), start() and toTable() should never be used in SDP code.
• --full-refresh-all cannot be combined with --full-refresh or --refresh.
• On Databricks, Lakeflow Pipelines add an Issues pane, event log details, Refresh failed tables and automatic retries for Jobs and API runs.

Why did my SDP flow fail?

In Spark Declarative Pipelines (SDP), a flow "reads data from a source, applies user-defined processing logic, and writes the result into a target dataset." When one fails, compare your project with the rules the Apache Spark programming guide documents. Start with a dry run, then work through the table below.

What is Spark Declarative Pipelines?

SDP is "a declarative framework for building reliable, maintainable, and testable data pipelines on Apache Spark" for batch and streaming data. Install it with pip install pyspark[pipelines] and import it in Python as from pyspark import pipelines as dp.
A pipeline project is a set of .py or .sql source files plus a YAML spec file, conventionally named spark-pipeline.yml or spark-pipeline.yaml. The spec requires name, libraries and storage:

name: my_pipeline
libraries:
  - glob:
      include: transformations/**
storage: file:///absolute/path/to/storage/dir

Datasets are streaming tables, materialized views and temporary views. A materialized view always has exactly one batch flow writing to it.

What are the common causes of a failed SDP flow?

Check	What the Spark guide says	What to do
Operations in dataset code	collect(), count(), pivot(), toPandas(), save(), saveAsTable(), start() and toTable() should never be used in SDP code. A dataset function must return a pyspark.sql.DataFrame	Keep only the code that defines the table or view
Spark session	SparkSession.builder.config(...).getOrCreate() mutates session config, which is blocked in SDP	Spark 4.1: declare spark = SparkSession.active(). Spark 4.2 and later: spark is injected
Dependency graph	A dry run reports graph validation errors such as cyclic dependencies	Remove a reference so no dataset depends on itself through a cycle
Sinks	Sinks support only streaming queries through append_flow, work only through the Python API, and are append-only	Target a sink with @dp.append_flow(target=...) over a streaming read, from Python
Refresh options	--full-refresh-all cannot be combined with --full-refresh or --refresh	Use one form per run

How do I catch errors before a run?

spark-pipelines dry-run "launches an execution of a pipeline that doesn't write or read any data, but catches many kinds of errors that would be caught if the pipeline were to actually run." The guide gives three examples: syntax errors (invalid Python or SQL), analysis errors (selecting from a table or column that doesn't exist) and graph validation errors (cyclic dependencies).
It accepts --spec; without it, the CLI searches the current directory and parent directories for the spec file.

spark-pipelines dry-run --spec /path/to/my-pipeline.yaml

What does a dataset function that breaks the rules look like?

The guide says SDP "evaluates the code that defines a pipeline multiple times during planning and pipeline runs," and that functions that define datasets should include only the code required to define the table or view.

from pyspark import pipelines as dp

@dp.materialized_view
def orders_bad():
    df = spark.table("orders")
    print(df.count())  # count() should never be used in SDP code
    return df

@dp.materialized_view
def orders_ok():
    return spark.table("orders")

How do I define several datasets in a for loop?

Pass the loop variable as a default argument, as the Spark guide's example does with region_filter=region. Python looks up a variable from the enclosing scope when the function is called, not when it is defined, so without the default every dataset function would see the last value (see the Python FAQ). The guide also requires the list you loop over to be "always additive".

How does Lakeflow Pipelines help me debug a failed update?

On Databricks, "Lakeflow pipelines extend and are interoperable with Spark Declarative Pipelines, while running on the performance-optimized Databricks Runtime." When an update fails, the docs say to start at the Issues pane in the pipeline editor or the bottom pane of the monitoring page, where a View logs button jumps to the relevant event log entries. The event log's error and details columns carry the full stack trace. Refresh failed tables retries only the failed tables plus anything downstream.
Retries depend on how an update starts. Updates started from Jobs or the Pipelines API, and all updates for continuous pipelines, use automatic retry and restart. Non-continuous updates started from the monitoring UI or the editor disable pipeline retries so you can see errors immediately.

Property	Default
pipelines.maxFlowRetryAttempts	Two retry attempts, so a flow can run three times including the original attempt
pipelines.numUpdateRetryAttempts	Five for triggered pipelines, unlimited for continuous pipelines

FAQs

Can I use pivot() in SDP?

The guide lists pivot() among the operations that should never be used in SDP code, and says the PIVOT clause is not supported in SDP SQL.

Why is old data still in my sink after a full refresh?

The guide says sinks are append-only: full refresh updates reset checkpoints but do not clean previously computed results, so earlier output stays at the destination.

How do I reset only some datasets?

Run spark-pipelines run --full-refresh orders,customers to clear all existing data and checkpoints for those datasets and recompute them from scratch. Use --refresh with a list of datasets to update them without clearing existing data, or --full-refresh-all to reset the whole graph. With no refresh option, SDP performs a default incremental update.