Dagster Spark

0.29.0 · active · verified Tue Apr 14

Dagster Spark is a Python library that provides integration components for orchestrating Apache Spark jobs within the Dagster data platform. It enables users to define, run, and monitor Spark-based data pipelines with Dagster's declarative programming model, offering capabilities for data management, lineage, and observability. The library is actively maintained and typically releases in sync with the core Dagster library.

Warnings

breaking The `SparkSolidDefinition` has been removed. Users should migrate to `create_spark_op` for defining Spark-based operations.
Fix: Replace `SparkSolidDefinition` with `create_spark_op`.
deprecated Spark Step Launchers are superseded by Dagster Pipes and are no longer the recommended method for launching external code from Dagster ops and assets. While still available, they will not receive new features or active development.
Fix: Consider migrating to Dagster Pipes for more lightweight and flexible Spark job orchestration, especially for new projects or feature development.
gotcha The Spark Declarative Pipeline (SDP) integration components (`SparkDeclarativePipelineComponent`, `SparkPipelinesResource`) are in feature preview. This API may have breaking changes in patch version releases and is not considered ready for production use.
Fix: Use with caution in production environments. Monitor Dagster release notes for updates on SDP API stability.
gotcha Ensure compatibility between your `dagster-spark` version, the `pyspark` library version (if used), and your Apache Spark cluster version. Specific Hadoop/AWS Java SDK versions might also be critical for integrations like S3.
Fix: Refer to Dagster's official documentation for recommended compatibility matrices. Test deployments thoroughly across environments.
breaking Dagster core (a dependency of `dagster-spark`) no longer supports Python 3.8 and requires `pydantic>=2`.
Fix: Upgrade your Python environment to 3.10 or newer and ensure `pydantic` is version 2 or greater.

Install

pip install dagster dagster-spark Install Dagster Spark

Imports

create_spark_op

from dagster_spark import create_spark_op

define_spark_config

from dagster_spark import define_spark_config

SparkDeclarativePipelineComponent
```
from dagster_spark.components.spark_declarative_pipeline import SparkDeclarativePipelineComponent
```
This API is currently in feature preview and not considered ready for production use; it may have breaking changes in patch releases.

Quickstart

This quickstart defines a Dagster job that executes a simple Spark application (like SparkPi) using `create_spark_op`. It demonstrates how to define Spark configuration and wrap a Spark job for orchestration within Dagster. Replace `path/to/your/spark-examples.jar` with the actual path to your Spark application JAR.

from dagster import job, Definitions, asset
from dagster_spark import create_spark_op, define_spark_config

# Define Spark configuration
my_spark_config = define_spark_config(
    {
        "spark.master": "local[*]",
        "spark.app.name": "dagster-spark-example"
    }
)

# Create an op from the Spark job definition
my_spark_job_op = create_spark_op(
    main_class="org.apache.spark.examples.SparkPi",
    jars=["path/to/your/spark-examples.jar"], # Replace with your Spark job JAR path
    spark_config=my_spark_config,
    name="my_spark_pi_op"
)

@job
def spark_pi_job():
    my_spark_job_op()

# Or, integrate with assets (requires PySparkResource from dagster-pyspark for direct SparkSession)
# For a simple asset that just runs a Spark job via `spark-submit`, you might do:
# @asset
# def spark_data_asset():
#    # This would involve using create_spark_op within an asset or using Dagster Pipes
#    # For simplicity, this example focuses on a job.
#    pass

defs = Definitions(jobs=[spark_pi_job])

view raw JSON →