Great question. In Prefect 3.x, subflows are most useful when you want to compose a larger workflow from well-defined stages and make each stage independently visible and diagnosable in the UI. When subflows shine: - Clear stage boundaries: e.g., extract, transform, load as separate runs you can click into. - Reuse: encapsulate a reusable workflow and call it with different parameters. - Different execution profiles: run different subflows with different task runners or settings. - Conditional orchestration: choose which subflow(s) to run based on logic. - Better observability: each subflow has its own logs, state history, and metrics. Docs: - Flows and subflows: Organize flows with subflows - run_deployment (for calling deployed flows from flows): API: run_deployment - Deployments overview: Deployments concepts Example: simple ETL with subflows (each stage visible in the UI)

from prefect import flow, task
from prefect.logging import get_run_logger

# Tasks
@task
def extract_task(source: str) -> list[str]:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Extracting from {source}")
    return [f"row-{i}" for i in range(3)]

@task
def transform_task(rows: list[str]) -> list[str]:
    logger = get_run_logger()
    <http://logger.info|logger.info>("Transforming rows")
    return [r.upper() for r in rows]

@task
def load_task(rows: list[str]) -> int:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Loading {len(rows)} rows")
    # pretend to load
    return len(rows)

# Subflows (each will appear as a separate flow run in the UI)
@flow(name="extract-stage", retries=1)
def extract_flow(source: str = "database") -> list[str]:
    return extract_task(source)

@flow(name="transform-stage")
def transform_flow(rows: list[str]) -> list[str]:
    return transform_task(rows)

@flow(name="load-stage")
def load_flow(rows: list[str]) -> int:
    return load_task(rows)

# Orchestrator (parent) flow
@flow(name="etl-orchestrator")
def etl_main(source: str = "database") -> int:
    # Each of these calls creates a nested subflow run with its own logs and state
    rows = extract_flow(source=source)
    processed = transform_flow(rows)
    loaded_count = load_flow(processed)

    logger = get_run_logger()
    <http://logger.info|logger.info>(f"ETL complete: loaded {loaded_count} rows")
    return loaded_count

if __name__ == "__main__":
    etl_main("postgres")

What you’ll see in the UI: - One run for etl-orchestrator (the parent). - Three subflow runs: extract-stage, transform-stage, load-stage. - In the parent’s graph, the subflow calls appear as special nodes linking to the child runs. - Each subflow has its own logs, state, and retries. Note: The above shows direct subflow calls (flow-from-flow). If you want to trigger a deployed flow from a flow (with options like timeouts or decoupling from the parent), use `run_deployment()`:

from prefect import flow
from prefect.deployments import run_deployment

@flow
async def orchestrator_calling_deployment():
    # "child-flow/deployment-name"
    child_run = await run_deployment(
        "child-flow/my-deployment",
        parameters={"param": "value"},
        timeout=None,        # wait until finished
        as_subflow=True      # link as subflow (default)
    )
    return child_run

If you have a specific use case (e.g., retries, cancellations, or different task runners across subflows), share a bit more detail and I can tailor the pattern and code for you.

Great use case for subflows. For a long, multi-step pipeline that you need to repeat across ~20 tables, I recommend: - Make each logical stage a subflow (so each stage is clickable/observable in the UI). - Keep the actual API calls and data moves as tasks (for retries, caching, and parallelism). - Drive everything from an orchestrator flow that: - Reads a list of table configs - Kicks off one “per-table” subflow per table - Limits concurrency to respect rate limits - Aggregates results High-level structure: - per-table subflow: ensure all SF activities exist (or update), create/ensure automation wiring, trigger, monitor, then load to warehouse - parent orchestrator: run that subflow for each table (optionally in parallel with a concurrency cap) Docs: - Subflows: Organize flows with subflows - run_deployment (for isolating long subflows): API: run_deployment Example: structure + concurrency caps, with stubbed Salesforce/warehouse calls ``` from dataclasses import dataclass from typing import List, Optional import time from prefect import flow, task from prefect.logging import get_run_logger from prefect.task_runners import ThreadPoolTaskRunner # ------------------------- # Config types # ------------------------- @dataclass class TableConfig: name: str source_object: str query: str destination_table: str # add any per-table specifics, e.g., file naming, partitions, etc. # ------------------------- # Low-level tasks (API calls / data ops) # These would wrap your Salesforce/warehouse SDK calls # ------------------------- @task(retries=3, retry_delay_seconds=5) def ensure_query_activity(cfg: TableConfig) -> str: logger = get_run_logger() logger.info(f"Ensuring query activity for {cfg.name}") # 1) lookup existing query activity by name # 2) create or update with cfg.query # return its ID time.sleep(0.1) return f"query-activity-{cfg.name}" @task(retries=3, retry_delay_seconds=5) def ensure_extract_activity(cfg: TableConfig, query_activity_id: str) -> str: logger = get_run_logger() logger.info(f"Ensuring extract activity for {cfg.name}") # create/update extract activity wired to query_activity_id time.sleep(0.1) return f"extract-activity-{cfg.name}" @task(retries=3, retry_delay_seconds=5) def ensure_file_transfer_activity(cfg: TableConfig, extract_activity_id: str) -> str: logger = get_run_logger() logger.info(f"Ensuring file transfer activity for {cfg.name}") # create/update file transfer activity wired to extract_activity_id time.sleep(0.1) return f"file-transfer-activity-{cfg.name}" @task(retries=3, retry_delay_seconds=5) def ensure_automation(cfg: TableConfig, query_activity_id: str, extract_activity_id: str, file_transfer_activity_id: str) -> str: logger = get_run_logger() logger.info(f"Ensuring automation for {cfg.name}") # create/update automation that chains the activities in the right order time.sleep(0.1) return f"automation-{cfg.name}" @task(retries=3, retry_delay_seconds=5) def trigger_automation(automation_id: str) -> str: logger = get_run_logger() logger.info(f"Triggering automation {automation_id}") # start/run the automation time.sleep(0.1) return f"run-{automation_id}" @task(retries=0) def monitor_automation(run_id: str, timeout_seconds: int = 1800, poll_seconds: int = 10) -> str: logger = get_run_logger() logger.info(f"Monitoring run {run_id}") start = time.time() # poll automation status until success/failure or timeout while True: # status = check_status(run_id) status = "Completed" # replace with actual check if status in {"Completed", "Failed"}:

logger.info(f"Run {run_id} finished with status={status}") return status if time.time() - start > timeout_seconds: raise TimeoutError(f"Run {run_id} did not finish in {timeout_seconds}s") time.sleep(poll_seconds) @task(retries=3, retry_delay_seconds=5) def load_to_warehouse(cfg: TableConfig) -> str: logger = get_run_logger() logger.info(f"Loading extracted data for {cfg.name} into {cfg.destination_table}") # e.g., read file delivered by file transfer and load to warehouse time.sleep(0.1) return f"loaded:{cfg.destination_table}" # ------------------------- # Per-table subflow (appears as its own run in the UI) # ------------------------- @flow(name="per-table-pipeline", retries=1, retry_delay_seconds=30) def per_table_pipeline(cfg: TableConfig) -> dict: logger = get_run_logger() logger.info(f"Starting pipeline for table {cfg.name}") query_id = ensure_query_activity(cfg) extract_id = ensure_extract_activity(cfg, query_id) xfer_id = ensure_file_transfer_activity(cfg, extract_id) automation_id = ensure_automation(cfg, query_id, extract_id, xfer_id) run_id = trigger_automation(automation_id) status = monitor_automation(run_id) if status != "Completed": raise RuntimeError(f"Automation for {cfg.name} ended with {status}") load_result = load_to_warehouse(cfg) logger.info(f"Completed pipeline for {cfg.name}: {load_result}") return { "table": cfg.name, "automation_status": status, "load_result": load_result, } # ------------------------- # Orchestrator flow # - Kicks off per-table subflows # - Limits concurrency with a task runner # ------------------------- @task def get_table_configs() -> List[TableConfig]: # Replace with reading a file/DB/config service return [ TableConfig( name="accounts", source_object="Account", query="SELECT Id, Name FROM Account", destination_table="dw.accounts", ), TableConfig( name="contacts", source_object="Contact", query="SELECT Id, Email FROM Contact", destination_table="dw.contacts", ), # ... add ~20 configs ] # Launcher task to enable parallel subflow runs @task def run_per_table(cfg: TableConfig) -> dict: # Calling a flow inside a task creates a subflow run and allows parallelism return per_table_pipeline(cfg) @flow(name="salesforce-batch-orchestrator", task_runner=ThreadPoolTaskRunner(max_workers=5)) def orchestrate_all_tables(parallelism: int = 5) -> List[dict]: logger = get_run_logger() configs = get_table_configs() # Submit per-table subflows via a launcher task to run them concurrently # max_workers on the ThreadPoolTaskRunner effectively caps concurrency here futures = [run_per_table.submit(cfg) for cfg in configs] results = [f.result() for f in futures] logger.info("All tables processed") return results if name == "__main__": orchestrate_all_tables() ``` Notes and options: - Concurrency: adjust ThreadPoolTaskRunner(max_workers=...) to respect API rate limits. For heavy I/O, threads are fine; for CPU-bound, consider Dask or Ray task runners. - Idempotency: each “ensure_…” task should first look up existing resources and update instead of re-creating. This keeps runs repeatable. - Retries: set sensible retries on API tasks; keep per-table subflow retry low so failures surface per table, not hidden by many retries. - Observability: each table run is a subflow with its own logs/status, so you can pinpoint which table failed. - Isolation alternative: if each table run is very long or you want independent retry/cancellation, create a deployment for the per-table subflow and trigger via

run_deployment(...)

from the orchestrator instead of calling it directly. That gives you timeouts and optional decoupling.

- API: run_deployment If you share a bit more about: - Are you using Salesforce Marketing Cloud vs Sales Cloud? - Your warehouse (Snowflake/BigQuery/Redshift/etc.) - Desired parallelism and any API rate limits I can tailor the tasks (auth, SDK usage, rate limiting, error handling) and deployment pattern to your environment.

Got it — we can keep your “data bus” pattern and layer Prefect on top. Below is a complete example that: - Keeps each step as a function that takes a dict and returns the same dict - Uses tasks for the work, a per-table subflow to group steps, and an orchestrator to run ~20 tables with a concurrency cap - Gives you clear subflow runs per table in the UI Docs for reference: - Subflows: Organize flows with subflows - Deployments (if you later want to isolate each per-table run): Deployments Example ``` import json from pathlib import Path from typing import Dict, List import pandas as pd from prefect import flow, task from prefect.logging import get_run_logger from prefect.task_runners import ThreadPoolTaskRunner # ------------------------- # Low-level "data bus" tasks # Each takes a dict and returns the same dict # ------------------------- @task(retries=2, retry_delay_seconds=5) def initialisation(data_bus: Dict) -> Dict: # Optionally enrich/validate bus; shown here reading defaults from a JSON file logger = get_run_logger() cfg_path = data_bus.get("config_path", "config.json") if Path(cfg_path).exists(): with open(cfg_path, "r") as f: defaults = json.load(f) # merge defaults into bus without clobbering explicit keys for k, v in defaults.items(): data_bus.setdefault(k, v) logger.info(f"Loaded defaults from {cfg_path}") return data_bus @task(retries=3, retry_delay_seconds=5) def extract_data(data_bus: Dict) -> Dict: logger = get_run_logger() src = data_bus["source_file_path"] logger.info(f"Extracting from {src}") df = pd.read_csv(src) data_bus["source_df"] = df return data_bus @task(retries=2, retry_delay_seconds=5) def transform_data(data_bus: Dict) -> Dict: # Put your transformations here; demo: add a simple column logger = get_run_logger() df = data_bus["source_df"] df["row_number"] = range(1, len(df) + 1) data_bus["source_df"] = df logger.info("Applied transformations") return data_bus @task(retries=3, retry_delay_seconds=5) def load_data(data_bus: Dict) -> Dict: logger = get_run_logger() out = data_bus["target_file_name"] data_bus["source_df"].to_csv(out, index=False) logger.info(f"Wrote {len(data_bus['source_df'])} rows to {out}") data_bus["load_result"] = {"rows": len(data_bus["source_df"]), "target": out} return data_bus # ------------------------- # Per-table subflow # Groups the steps; appears as its own run in the UI # ------------------------- @flow(name="per-table-pipeline", retries=1, retry_delay_seconds=15) def per_table_pipeline(data_bus: Dict) -> Dict: logger = get_run_logger() logger.info(f"Starting pipeline for table_key={data_bus.get('table_key','unknown')}") data_bus = initialisation(data_bus) data_bus = extract_data(data_bus) data_bus = transform_data(data_bus) data_bus = load_data(data_bus) logger.info(f"Completed table_key={data_bus.get('table_key','unknown')}") return data_bus # ------------------------- # Orchestrator # Loads ~20 table configs and runs per-table subflows with a concurrency cap # ------------------------- @task def load_table_buses() -> List[Dict]: # In practice, load from a config store/DB; here we stub 2 entries return [ { "table_key": "accounts", "source_file_path": "data/accounts.csv", "target_file_name": "out/accounts_loaded.csv", # optional: "config_path": "accounts_config.json" }, { "table_key": "contacts", "source_file_path": "data/contacts.csv", "target_file_name": "out/contacts_loaded.csv", }, # ... add ~20 more ] # launcher task to enable parallel subflow runs @task def run_one_table_bus(data_bus: Dict) -> Dict:

# Calling a flow inside a task creates a subflow run and enables parallelism when submitted return per_table_pipeline(data_bus) @flow(name="batch-orchestrator", task_runner=ThreadPoolTaskRunner(max_workers=5)) def orchestrate_all_tables() -> List[Dict]: logger = get_run_logger() buses = load_table_buses() # Kick off multiple per-table subflows concurrently (capped by max_workers) futures = [run_one_table_bus.submit(bus) for bus in buses] results = [f.result() for f in futures] # Aggregate/inspect results if needed total_rows = sum(r.get("load_result", {}).get("rows", 0) for r in results) logger.info(f"Processed {len(results)} tables; total rows={total_rows}") return results if name == "__main__": orchestrate_all_tables() ``` Notes to keep this pattern robust in Prefect: - Keep returning the updated dict from every task/flow. This keeps dependencies explicit in the DAG and makes retries predictable. - If you deploy and run on remote workers, passing big objects (like large DataFrames) across subflow boundaries can be expensive due to serialization. In those cases, consider writing to an object store and passing a path/URI in the data bus instead of the DataFrame itself. - Use retries on the API/file I/O tasks rather than on the whole per-table subflow, so failures isolate to the smallest unit. - Concurrency and rate limits: adjust

ThreadPoolTaskRunner(max_workers=...)

. If you need stricter service-wide limits, we can add Prefect concurrency limits as well. If you share how your per-table configs are stored and whether these will run as a deployment on workers, I can tailor the data bus contents (e.g., add storage URIs, auth handles) and the parallelism pattern accordingly.

Thanks for the details — a couple of key 2.8.0 specifics are tripping you up: - ConcurrentTaskRunner exists in Prefect 2.8.0, but it does not accept max_workers. Passing it will raise “unexpected keyword argument 'max_workers'”. See source: ConcurrentTaskRunner (2.8.0) - Flows cannot be run from within tasks in Prefect 2.x. Your task run_one_table_view calls a flow per_table_pipeline, which will raise: “Flows cannot be run from within tasks. Did you mean to call this flow in a flow?”. See source: engine restriction Recommended 2.8.0 pattern for your “data bus” design - Keep each step as a task that takes and returns the dict. - In the parent flow, build a per-view chain of task submissions (init -> extract -> …) and do that for each view. This gives you per-view sequencing while letting multiple views run concurrently with the default ConcurrentTaskRunner. - If you need to cap concurrency, do simple batching (since ConcurrentTaskRunner has no max_workers in 2.8.0). For more control, consider DaskTaskRunner from prefect-dask. Working example (2.8.0-compatible) ``` import json from pathlib import Path from typing import Dict, List from datetime import datetime import pandas as pd from prefect import flow, task from prefect.logging import get_run_logger from prefect.task_runners import ConcurrentTaskRunner from typing import Literal # Optional: GCP logger helper def log_info(data_bus: Dict, priority: str, message: str) -> None: payload = { "timestamp": datetime.now().strftime("%Y/%m/%d %H:%M:%S"), "Pipeline": data_bus.get("log", {}).get("pipeline"), "stage": data_bus.get("log_stage"), "message": message, } # requires data_bus["logger"] to be a configured GCP logger if "logger" in data_bus: data_bus["logger"].log_struct(payload, severity=priority) # ------------------------- # Data-bus tasks # ------------------------- @task(retries=2, retry_delay_seconds=5) def initialisation(view: Dict, run_type: Literal["DEV", "PROD"]) -> Dict: config = "configs/dev_config.json" if run_type == "DEV" else "configs/prod_config.json" with open(config, "r") as f: data_bus = json.load(f) data_bus["data_views"] = view data_bus["log_stage"] = "initialisation" return data_bus @task(retries=3, retry_delay_seconds=5) def extract_data(data_bus: Dict) -> Dict: logger = get_run_logger() logger.info(f"Extract stub for view={data_bus['data_views'].get('name','unknown')}") data_bus["log_stage"] = "extract" # Example: attach a DataFrame or path to extracted data data_bus["source_df"] = pd.DataFrame([{"a": 1}, {"a": 2}]) return data_bus @task(retries=2, retry_delay_seconds=5) def transform_data(data_bus: Dict) -> Dict: logger = get_run_logger() logger.info("Transform stub") data_bus["log_stage"] = "transform" df = data_bus["source_df"] df["row_number"] = range(1, len(df) + 1) data_bus["source_df"] = df return data_bus @task(retries=3, retry_delay_seconds=5) def load_data(data_bus: Dict) -> Dict: logger = get_run_logger() out = data_bus.get("target_file_name", "out/default.csv") Path(out).parent.mkdir(parents=True, exist_ok=True) data_bus["source_df"].to_csv(out, index=False) data_bus["log_stage"] = "load" data_bus["load_result"] = {"rows": len(data_bus["source_df"]), "target": out} logger.info(f"Wrote {data_bus['load_result']['rows']} rows to {out}") return data_bus # ------------------------- # Orchestrator flow # - Builds per-view chains of task submissions # - Multiple chains run concurrently under ConcurrentTaskRunner # - Optional batching to cap concurrency # ------------------------- def load_table_views() -> List[Dict]: with open("configs/data_views_complete.json", "r") as f: return json.load(f)

@flow(name="batch-orchestrator", task_runner=ConcurrentTaskRunner()) def orchestrate_all_tables(run_type: Literal["DEV", "PROD"], batch_size: int = 0) -> List[Dict]: logger = get_run_logger() views = load_table_views() def run_batch(batch_views: List[Dict]) -> List: # Build chains: init -> extract -> transform -> load per view last_step_futures = [] for view in batch_views: init_f = initialisation.submit(view, run_type) extract_f = extract_data.submit(init_f) transform_f = transform_data.submit(extract_f) load_f = load_data.submit(transform_f) last_step_futures.append(load_f) # Wait for batch to finish return [f.result() for f in last_step_futures] results: List[Dict] = [] if batch_size and batch_size > 0: # Process in batches to limit concurrency for i in range(0, len(views), batch_size): batch = views[i : i + batch_size] logger.info(f"Processing batch {i//batch_size + 1} with {len(batch)} views") results.extend(run_batch(batch)) else: # Fire all chains concurrently (be mindful of API limits) results = run_batch(views) logger.info(f"Processed {len(results)} views total") return results if name == "__main__": orchestrate_all_tables(run_type="DEV", batch_size=5) # set batch_size=0 to remove cap ``` Why this works in 2.8.0 - No flow-inside-task calls (which 2.x forbids). - Concurrency comes from submitting many independent per-view task chains within a flow. The ConcurrentTaskRunner (default when using .submit) executes them concurrently. - Per-view steps still run in order because each step depends on the prior step’s future. If you really want subflows per view - In 2.8.0, subflows called from flows are easy but run sequentially by default. Concurrent subflows in 2.8 had some quirks; if you need true concurrent subflows, consider upgrading to Prefect ≥2.10 or to 3.x, or trigger each view via deployments (each as its own flow run) rather than nested subflows. Optional: more control over concurrency - Dask: install prefect-dask and use DaskTaskRunner to set worker counts and limits. - Batching (shown) is a simple, robust cap with the built-in runner. If you hit another error after removing max_workers and restructuring the concurrency as above, paste the traceback and I’ll pinpoint it.