https://prefect.io
Join Slack
Hello everyone, I'm using `start_flow_run.map(...)` to use the same flow, but iterate across some pr...
# ask-community
j
Hello everyone, I'm using 
start_flow_run.map(...)
to use the same flow, but iterate across some provided parameters. Unfortunately when I set the 
flow_name=unmapped('webfs')
I get this error when attempting to run the flow: `ValueError: Received both 
flow_id
and 
flow_name
. Only one flow identifier can be passed.`
✅ 1
z
Hey @Joe -- just to check, you're not passing a 
flow_id
too?
j
Negative
Copy code
storage = create_storage(flow_name=flow_name)
    run_config = UniversalRun(labels=['local'])

    with Flow(flow_name, storage=storage, run_config=run_config, result=storage.result) as flow:
        image = parameter_webfs_image()
        command = parameter_container_command()
        network = parameter_container_network()

        create_host_config = CreateHostConfigWithMount()
        create_container = CreateContainer(trigger=not_all_skipped)
        start_container = StartContainer()
        mongo_volumes = MongoVolumes()
        volumes = mongo_volumes()
        mapped_flows = create_flow_run.map(
            volumes,
            flow_name=unmapped('webfs'),
            project_name=unmapped(project_name),
        )
z
You're passing 
volumes
as a positional argument
mm
Yeah. The first argument is 
flow_id
so you're passing it in there.
j
Ah, I'm unclear on how to pass the mapped information in that case... Is it another wrapper?
like the unmapped one I mean
z
Perhaps you want...
Copy code
mapped_flows = create_flow_run.map(
            parameters={"volume": volumes},
            flow_name=unmapped('webfs'),
            project_name=unmapped(project_name),
        )
j
and that will iterate across the volumes?
z
I think so, I'm actually not sure with it wrapped in a dictionary like that
That might require something else..
But generally yes, any task inputs you pass that aren't wrapped in 
unmapped
are mapped over.
@Kevin Kho I think you'll know how to do this better than I
j
Ahhh, OK now I see
That does appear to do it since my volumes list is just the dict for task parameters
Thanks!
z
You're welcome!
k
Was this fixed @Joe? Just got here 😅
j
Yeah - I was having a hard time interpreting the docstring for how to pass the iterated argument(s) not realizing the goal was to pass any **kwarg that the mapped task accepted as an iterable of the arguments valid type(s)
Which gets very abstract once you start executing flows as tasks due to the massive parameter space.
20 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by