Short answer: Prefect orchestrates Spark, it doesn’t create a Spark cluster for you. Don’t try to “spin up a temporary Spark cluster” via the ECS worker. Instead, point Prefect at a Spark runtime you control (Databricks, EMR/EMR Serverless, AWS Glue, or an existing Spark cluster) and submit jobs there. Given you’re on AWS and already planning an ECS work pool on EC2, here are the practical options: - Easiest serverless/ephemeral: AWS Glue - Prefect has a first-class Glue block you can call from flows; Glue provisions Spark behind the scenes per run and tears it down automatically. - Good fit if you want per-job ephemeral Spark without managing clusters. - Docs: prefect-aws collection and Glue job block - Also ephemeral: EMR Serverless - No dedicated Prefect block today, but you can call

boto3

from a Prefect task to

StartJobRun

and poll for completion. - Good if you need EMR-compatible runtime or finer control than Glue. - AWS docs: What is EMR Serverless? - Managed platform: Databricks - Mature Prefect integration for Spark jobs (run existing job, submit-and-wait, get outputs). - Docs: prefect-databricks - Existing/long-lived Spark cluster (EMR on EC2, self-managed on EC2, or Kubernetes) - Prefect can submit via

boto3

(EMR steps) or

spark-submit

(shell) to your cluster. About ECS: Prefect’s ECS worker runs your flows in ECS tasks. It does not orchestrate a multi-node Spark cluster for you. You could run a single-container Spark job (local mode) in ECS, or a driver that targets a separate cluster, but building and tearing down a proper ephemeral Spark cluster across multiple ECS tasks is non-trivial and not something Prefect handles out of the box. Code-first examples 1) AWS Glue (serverless Spark, recommended for ephemeral)

# pip install prefect-aws

from prefect import flow
from prefect.blocks.system import Secret
from prefect_aws import AwsCredentials
from prefect_aws.glue_job import GlueJobBlock

@flow
async def run_spark_via_glue():
    # Load AWS creds and DB creds stored in Prefect blocks
    aws = AwsCredentials.load("aws-creds-ap-southeast-1")
    db_user = Secret.load("rds-user").get()
    db_pass = Secret.load("rds-pass").get()

    # Arguments your Glue script expects (e.g., JDBC to RDS)
    glue = GlueJobBlock(
        job_name="my-glue-spark-job",  # pre-created Glue job
        arguments={
            "--jdbc_url": "jdbc:postgresql://<rds-endpoint>:5432/<db>",
            "--db_user": db_user,
            "--db_pass": db_pass,
        },
        aws_credentials=aws,
        job_watch_poll_interval=30.0,
    )

    run = await glue.trigger()
    run.wait_for_completion()  # raises on failure
    return "SUCCEEDED"

if __name__ == "__main__":
    import asyncio
    asyncio.run(run_spark_via_glue())

Notes for your setup: - Put Glue in the same VPC/subnets as RDS and allow SG-to-SG access so the job can reach the database. - Package your Glue script and any dependencies in S3; assign an execution role with S3/CloudWatchLogs permissions. 2) EMR Serverless (ephemeral EMR runtime) ``` # pip install prefect-aws boto3 from prefect import flow, task from prefect.blocks.system import Secret from prefect_aws import AwsCredentials import time @task def start_emr_serverless_job(app_id: str, role_arn: str, entry_point: str, spark_params: str, aws: AwsCredentials) -> str: client = aws.get_boto3_session().client("emr-serverless") resp = client.start_job_run( ApplicationId=app_id, ExecutionRoleArn=role_arn, JobDriver={ "SparkSubmit": { "EntryPoint": entry_point, # e.g., s3://bucket/jobs/my_job.py

"SparkSubmitParameters": spark_params # e.g., "--conf spark.executor.memory=4g ..." } }, Name="prefect-spark-job", ) return resp["JobRunId"] @task def wait_for_emr_serverless_job(app_id: str, job_run_id: str, aws: AwsCredentials): client = aws.get_boto3_session().client("emr-serverless") terminal = {"SUCCESS", "FAILED", "CANCELLING", "CANCELED"} while True: s = client.get_job_run(ApplicationId=app_id, JobRunId=job_run_id)["JobRun"]["State"] if s in terminal: if s != "SUCCESS": raise RuntimeError(f"EMR Serverless job ended with state: {s}") return s time.sleep(15) @flow def run_spark_on_emr_serverless(): aws = AwsCredentials.load("aws-creds-ap-southeast-1") db_user = Secret.load("rds-user").get() db_pass = Secret.load("rds-pass").get() app_id = "<emr-serverless-app-id>" role_arn = "<iam-role-for-emr-serverless>" entry_point = "s3://bucket/jobs/my_spark_job.py" spark_params = f"--conf spark.executor.instances=2 --conf spark.executor.memory=4g --db_user {db_user} --db_pass {db_pass}" job_run_id = start_emr_serverless_job(app_id, role_arn, entry_point, spark_params, aws) return wait_for_emr_serverless_job(app_id, job_run_id, aws)

Notes:
- Put the EMR Serverless app in the same VPC/subnets as RDS and allow SG access.
- Package your code to S3; give the app execution role S3/Logs permissions.

3) Submitting to an existing Spark cluster (spark-submit)

# pip install prefect-shell from prefect import flow from prefect_shell import shell_run_command @flow def run_spark_submit(): # Assumes spark-submit is available in PATH inside the worker environment res = shell_run_command( command=[ "spark-submit", "--master", "spark//&lt;master host&gt;7077", "--deploy-mode", "client", "--class", "com.example.App", "s3://bucket/jars/app.jar", "--some-arg", "value", ], stream_output=True, ) return res ``` What I recommend for your setup in ap-southeast-1 - If you want ephemeral, managed Spark with minimal ops: use Glue or EMR Serverless. Glue is the most “turnkey” with a Prefect block. - If you already use Databricks or want its features: use the Databricks integration. - I would not try to build a temporary multi-node Spark cluster via Prefect’s ECS worker. It’s possible but complex (you’d have to create ECS services/tasks for master and executors, set up service discovery/networking, coordinate lifecycle, then tear down). Questions to tailor this for you: - Are you OK using AWS Glue or EMR Serverless for ephemeral Spark? - Do you already have Spark code packaged (Py files, JAR/wheel) and a place in S3? - Will the Spark job need VPC access to your RDS (sounds like yes)? If so, we’ll confirm subnets/SGs and JDBC connectivity details.

Great question. Prefect doesn’t have a single “Repair run” button like Databricks, but you can get the same effect: on a retry, only the failed task(s) and anything downstream will re-execute, while previously successful tasks are skipped automatically. The key is task result caching. How to achieve Databricks-style “repair” in Prefect - Enable result persistence and caching for your tasks so successful steps are automatically skipped on reruns. - Retry the failed flow run. Prefect will only re-run tasks that need work (failed or whose inputs changed); everything else returns from cache and is marked Cached. What to set up 1) Persist results and cache by inputs (typical choice)

from prefect import task, flow
from prefect.cache_policies import INPUTS

@task(persist_result=True, cache_policy=INPUTS)
def step_n(...):
    ...

- persist_result=True ensures the task’s output is stored. - cache_policy=INPUTS means reusing results when inputs are identical across retries. You can also set persistence globally:

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

2) Retry the failed flow run - In UI: open the failed flow run and click Retry. - CLI:

prefect flow-run retry <flow-run-id-or-name>

On retry, Prefect: - Skips all tasks with valid cached results (shows state “Cached”) - Re-executes the failed task and any downstream tasks whose inputs depend on it Why this matches “repair” - Databricks “Repair” runs only the failed nodes and their dependents. - Prefect achieves the same outcome via caching: upstream tasks with unchanged inputs don’t re-run; downstream of the failed task re-runs because their inputs changed (or because the failed step had no cached result). Extra controls and tips - If you want to re-run from a specific point even if caches exist, set refresh_cache=True for the task(s) you want to force:

t3 = step_3.with_options(refresh_cache=True)(...)

- If you want cache to expire (e.g., recompute after a day):

from datetime import timedelta

@task(persist_result=True, cache_policy=INPUTS, cache_expiration=timedelta(days=1))
def step(...): ...

- For very large pipelines, consider transactions to ensure “all-or-nothing” caching across a block of steps. Docs and references - How to retry failed runs: https://docs.prefect.io/v3/how-to-guides/workflows/retry-flow-runs/ - Cache workflow steps: https://docs.prefect.io/v3/how-to-guides/workflows/cache-workflow-steps/ - Caching concepts: https://docs.prefect.io/v3/concepts/caching - Advanced results/persistence: https://docs.prefect.io/v3/advanced/results If you share a quick shape of your 50-step graph (where the failure happens and how downstream depends on it), I can suggest cache policies per step to maximize “repair”-style behavior on retries.