Create Production-Ready Prefect Workflows

This skill creates production-ready Prefect 2.x workflow orchestration flows. Core principles cover task granularity (logical, independently retryable units), idempotency, strategic error boundaries with Prefect's retry mechanism, careful resource handling, and comprehensive logging via get_run_logger(). A basic ETL flow decorates extraction with @task(retries=3, retry_delay_seconds=60), submits extraction tasks concurrently across multiple sources using a ConcurrentTaskRunner, gathers futures, transforms the combined data (skipping and logging individual record failures rather than failing the whole batch), and loads the result.

@flow(task_runner=ConcurrentTaskRunner())
def etl_pipeline(sources: List[str], destination: str):
    extraction_futures = [extract_data.submit(source, "SELECT * FROM table") for source in sources]
    all_data = [f.result() for f in extraction_futures]

Advanced task patterns cover conditional execution (a freshness-check task gating whether an update task actually runs, returning a Completed state with a message when skipped) and dynamic task generation (submitting one task per item in a runtime-determined list, then aggregating all results). Error handling and monitoring differentiate exception types explicitly - validation errors (ValueError) are raised without retry, connection errors are raised to trigger Prefect's retry mechanism, and unexpected exceptions are logged with full traceback - with a flow-level on_failure hook sending a Slack notification via a configured SlackWebhook block. Configuration management uses Prefect Blocks (Secret, JSON) to load credentials and config securely rather than hardcoding them, and deployment is defined via Deployment.build_from_flow() with a cron schedule, target work queue, default parameters, and tags. Performance optimization covers choosing ConcurrentTaskRunner for I/O-bound work vs. DaskTaskRunner for compute-intensive parallel processing, caching expensive task results via cache_key_fn, using task.map() for efficient collection processing, and setting resource limits/timeouts. Testing uses prefect_test_harness() as a session-scoped fixture to run flows in an isolated test environment and assert on completion state and results.

Use this skill when building or hardening a Prefect data workflow - orchestrating concurrent or dynamically generated tasks, implementing retry-aware error handling with differentiated exception types, securely managing configuration via Blocks, scheduling production deployments, or testing flows with the Prefect test harness.

It is not the right tool for orchestration platforms other than Prefect (Airflow, Dagster, Temporal) since the flow/task decorator model and Blocks system are Prefect-specific, or for simple linear scripts with no need for retries, concurrency, or observability that a plain function would handle just as well.

Input: the data pipeline's extract/transform/load steps, their concurrency and retry requirements, and the deployment schedule needed. Output: a Prefect flow with decorated, retryable tasks, concurrent or dynamic task orchestration, differentiated exception handling with failure notifications, Blocks-based secure configuration, a scheduled deployment definition, and test-harness-based flow tests.

Built on Prefect 2.x (flow, task, ConcurrentTaskRunner, DaskTaskRunner, Deployment, Blocks like Secret/JSON/SlackWebhook), with pytest and prefect.testing.utilities.prefect_test_harness for testing.

Data engineers building production Prefect workflows - particularly those needing concurrent or dynamic task orchestration, retry-aware error handling with alerting, and scheduled, securely configured deployments.