Assuming you’re on Prefect 3.x: give each mapped task run a custom name by setting

task_run_name

on the task (or overriding it with

.with_options(...)

) before calling

.map

.

task_run_name

can be: - a string template using parameter names, or - a callable that returns a string (it can accept a

parameters

dict for per-item names) Examples 1) String template (simplest)

from prefect import flow, task

@task(task_run_name="process_{item}")
def process(item: str) -> str:
    return item.upper()

@flow
def my_flow():
    futs = process.map(item=["a", "b", "c"])
    return futs.result()

Each mapped run is named: process_a, process_b, process_c 2) Override per call with .with_options(...)

futs = process.with_options(task_run_name="special_{item}").map(item=["a", "b", "c"])

3) Callable using parameters (most flexible)

from prefect import flow, task

def name_from_params(parameters: dict) -> str:
    return f"user_{parameters['user_id']}"

@task(task_run_name=name_from_params)
def process_user(user_id: int) -> str:
    return f"ok:{user_id}"

@flow
def my_flow():
    futs = process_user.map(user_id=[101, 102, 103])
    return futs.result()

Run names: user_101, user_102, user_103 Notes - Templates reference task parameter names (and can handle nested dict keys). - You can also use a no-arg callable for a static name, but per-item naming typically uses the

parameters

callback or string templates.

Yes — that’s the right idea. In Prefect 3.x you can name each mapped run using a string template that references the task parameter names. For your example:

from prefect import task, flow

@task(task_run_name="process_{org_id}")
def start_etl_for_file_store(org_id: str) -> str:
    ...

@flow
def run():
    organizations = ["acme", "globex", "initech"]
    return start_etl_for_file_store.map(org_id=organizations).result()

Each mapped task run will be named like: process_acme, process_globex, process_initech. Notes - Templates use Python’s .format(), so you can also do nested access like

{org[id]}

if

org

is a dict, or

{org.name}

if it’s an object. - You can override per call if needed:

start_etl_for_file_store.with_options(task_run_name="etl_{org_id}").map(org_id=organizations)