Spark Declarative Pipelines Dagster Integration

[michalcabir-ui]

Summary & Motivation

This PR introduces the SparkDeclarativePipelineComponent to the dagster-spark library, providing native orchestration and observability for the new Spark Declarative Pipelines (SDP) framework.

Key Features & Architectural Highlights:

Auto-Discovery: Automatically infers datasets (Materialized Views, Streaming Tables) via spark-pipelines dry-run, with a robust, regex-based static source parsing fallback for environments where dry-run is unavailable.

State Management: Implemented as a StateBackedComponent to cache discovered datasets, avoiding JVM overhead during daemon reloads.

Optimized Execution: Uses context.is_subset to intelligently pass explicit dataset keys or utilize the --full-refresh-all flag to avoid OS argument length limits.

Real-time Observability: Streams execution logs directly to Dagster's UI using unbuffered subprocess execution (PYTHONUNBUFFERED=1). Prevents memory leaks on long-running jobs by utilizing a bounded collections.deque for log tailing.

Strict Error Handling: Differentiates between discovery errors and execution failures (SparkPipelinesExecutionError), preventing silent exception swallowing during source parsing.

How I Tested These Changes

Added 22 comprehensive unit tests across test_component.py, test_discovery.py, and test_resource.py.

Verified complex JSON and structured text parsing fallbacks, including deduplication validation.

Tested subprocess execution paths, ensuring accurate log streaming, timeout handling (configured to 60s for daemon safety), and proper exception bubbling.

Validated multi-asset generation, ensuring temporary views are filtered out by default unless explicitly overridden.

Changelog

dagster-spark: Added SparkDeclarativePipelineComponent for orchestrating Spark Declarative Pipelines (SDP).

[greptile-apps]

Greptile Summary

This PR introduces SparkDeclarativePipelineComponent, a StateBackedComponent that orchestrates Spark Declarative Pipelines (SDP) within Dagster. It auto-discovers datasets via spark-pipelines dry-run (with a regex-based static fallback), caches them as serialized SparkPipelineState, and exposes them as a can_subset=True multi_asset backed by SparkPipelinesResource. Prior review concerns (subprocess leak in run_and_observe and silent config-type swallowing in the resolver) are both addressed. The implementation is generally solid and well-tested, with 22 unit tests across component, discovery, and resource layers.

A few non-blocking issues remain:

• SparkPipelinesResource is not re-exported from dagster_spark/__init__.py, meaning from dagster_spark import SparkPipelinesResource raises an ImportError for users who need to configure it directly.
• ExecutionMode is duplicated in both component.py and resource.py; consolidating into a shared location avoids the two definitions drifting.
• discover_datasets default mode (dry_run_only) differs from the component default (dry_run_with_fallback), which may surprise users who call the resource method directly and get strict failure-on-error behaviour instead of the graceful fallback they'd expect.
• Regex patterns are reconstructed on every iteration of the outer loop in _extract_report_text; hoisting them above the loop is a minor performance improvement.
• One test assertion in test_resource.py assumes the dataset string is always the final CLI argument, which becomes incorrect when run_extra_args is populated.

Confidence Score: 4/5

• PR is safe to merge; all previously raised blocking concerns are resolved and remaining issues are non-blocking style/discoverability improvements.
• Prior P0/P1 concerns (subprocess leak, silent config fallback) are addressed. Remaining comments are P2 suggestions — a missing top-level export, duplicated type alias, a default-mode inconsistency, a minor regex inefficiency, and one fragile test assertion. None of these block correct operation of the component in production.
• dagster_spark/__init__.py (missing SparkPipelinesResource export) and resource.py (duplicate ExecutionMode, default mode inconsistency) warrant a quick look before merge.

Important Files Changed

Filename	Overview
python_modules/libraries/dagster-spark/dagster_spark/components/spark_declarative_pipeline/component.py	Core component implementing StateBackedComponent with dataset filtering, path resolution, and multi_asset generation. The silent-fallback issue raised in prior review is now fixed. Minor: ExecutionMode is duplicated from resource.py.
python_modules/libraries/dagster-spark/dagster_spark/components/spark_declarative_pipeline/resource.py	subprocess resource leak is fixed with try/finally. ExecutionMode type alias duplicated from component.py. discover_datasets default mode (dry_run_only) differs from component default (dry_run_with_fallback), which may surprise callers.
python_modules/libraries/dagster-spark/dagster_spark/components/spark_declarative_pipeline/discovery.py	Comprehensive discovery logic with JSON+text fallbacks, deduplication guard, and source-scanning. _dataset_id_pattern and pattern tuples are recreated on every line in _extract_report_text's outer loop — minor performance issue.
python_modules/libraries/dagster-spark/dagster_spark/init.py	SparkPipelinesResource is not re-exported at the package level, making it harder to discover for users who want to configure it in their Dagster projects.
python_modules/libraries/dagster-spark/dagster_spark_tests/components/spark_declarative_pipeline/test_resource.py	Good coverage of run_and_observe paths. One assertion assumes dataset string is always the last CLI argument, which breaks if run_extra_args is populated.
python_modules/libraries/dagster-spark/dagster_spark_tests/components/spark_declarative_pipeline/test_discovery.py	Thorough tests covering JSON parsing, text fallback, noisy stdout, and duplicate-name detection. No significant issues.
python_modules/libraries/dagster-spark/dagster_spark_tests/components/spark_declarative_pipeline/test_component.py	Good lifecycle tests covering sandbox loading, state refresh, temporary view filtering, and override behaviour. No significant issues.
python_modules/libraries/dagster-spark/dagster_spark/components/spark_declarative_pipeline/scaffolder.py	Minimal scaffolder that writes a sensible default defs.yaml. No issues.
python_modules/libraries/dagster-spark/pyproject.toml	Correct entry-point registration for dg CLI. No issues.

Sequence Diagram

sequenceDiagram
    participant DG as Dagster Daemon
    participant Comp as SparkDeclarativePipelineComponent
    participant Disc as discover_datasets_fn
    participant CLI as spark-pipelines CLI
    participant State as StateBackedComponent (fs)
    participant Asset as _spark_pipeline_asset
    participant Res as SparkPipelinesResource

    Note over DG,State: Load / Refresh State Phase
    DG->>Comp: refresh_state(working_dir)
    Comp->>Disc: discover_datasets_fn(pipeline_spec_path, discovery_mode)
    alt dry_run_only / dry_run_with_fallback
        Disc->>CLI: spark-pipelines dry-run --spec ...
        CLI-->>Disc: JSON or text stdout
        Disc->>Disc: parse_dry_run_output_to_datasets()
    else fallback / source_only
        Disc->>Disc: discover_datasets_from_sources() (regex scan .py/.sql)
    end
    Disc-->>Comp: list[DiscoveredDataset]
    Comp->>State: serialize_value(SparkPipelineState) → state_path

    Note over DG,Asset: Build Defs Phase (daemon reload)
    DG->>Comp: build_defs_from_state(context, state_path)
    Comp->>State: deserialize_value(state_path) → SparkPipelineState
    Comp->>Comp: filter temporary_view datasets
    Comp->>Comp: build AssetSpecs via get_asset_spec()
    Comp-->>DG: Definitions(multi_asset, SparkPipelinesResource)

    Note over Asset,CLI: Execution Phase (materialize)
    Asset->>Res: run_and_observe(context, pipeline_spec_path, execution_mode, asset_keys)
    Res->>CLI: spark-pipelines run --spec ... [--refresh|--full-refresh|--full-refresh-all] [datasets]
    loop stdout streaming
        CLI-->>Res: log line
        Res->>DG: context.log.info(line)
    end
    CLI-->>Res: returncode
    alt returncode != 0
        Res-->>Asset: raise SparkPipelinesExecutionError
    else success
        Res-->>Asset: return
        Asset-->>DG: yield MaterializeResult(asset_key) per selected key
    end

Loading

_{Reviews (2): Last reviewed commit: "pyright fix" | Re-trigger Greptile}

[xionon]

Overall I think the most important thing here is that we need some kind of verification against an actual pyspark or spark server; an integration test, or a script to replay a dry run that exports some test fixtures.

[cmpadden]

@greptileai can you please update your review?

[cmpadden]

Thank you @michalcabir-ui.

Going to land this, and we can figure out if there are any tweaks / subsequent revisions to follow.

Comments have been addressed.