How to map over flows with various parameter values?

[anna_geller]

View in #prefect-community on Slack

@Jean-Baptiste_Six: Hi !
I have the following flow :

with Flow("sec daily", state_handlers=[flow_handler], result=GCSResult(BUCKET_RESULTS)) as sec_daily_flow:
    root_dir = init_dirs("sec", archive=False)
    download_dir = download(root_dir, False)
    metadatas = metadata(download_dir)
    upload(metadatas)
    result_locations = split_store_batches(metadatas)
    mapped_flow_run_ids = create_flow_run.map(
        flow_name=unmapped("index_flow"),
        project_name=unmapped("Document Pipeline"),
        parameters=result_locations
    )

With a “subflow”:

with Flow("index_flow", result=GCSResult(BUCKET_RESULTS), state_handlers=[flow_handler]) as index_flow:
    # Imports
    metadatas_batch_location = Parameter("metadatas_batch_location", required=True)
    metadatas_batch = get_result(metadatas_batch_location)
    imports = build_import(metadatas_batch)
    index_weaviate(imports)

But I have the following error :

prefect.exceptions.ClientError: [{'path': ['create_flow_run'], 'message': 'dictionary update sequence element #0 has length 1; 2 is required', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

That I’m not sure to understand I need some help please

Same error for both ‘batches’:
• Task ‘create_flow_run[0]’: Exception encountered during task execution!
• Task ‘create_flow_run[1]’: Exception encountered during task execution!

@Anna_Geller: Can you share the definition of split_store_batches?

@Jean-Baptiste_Six: Sure

@task(max_retries=1, retry_delay=timedelta(seconds=300), state_handlers=[fail_handler], log_stdout=True)
def split_store_batches(metadatas: List[dict]) -> List[str]:
    metadatas_batches = [{"metadatas_batch": batch_value} for batch_value in list(batch(metadatas, BATCH_SIZE_INDEX))]
    result_locations = []
    with tempfile.TemporaryDirectory() as tmp_dir:
        for metadatas_batch in metadatas_batches:
            today = prefect.context.get("today").split("-") # type: ignore
            file_id = uuid.uuid4()
            path_file = f"{tmp_dir}/{file_id}"
            result_location = f"{today[0]}/{today[1]}/{today[2]}/metadata_batches/{file_id}"
            with open(path_file, "w") as file:
                json.dump(metadatas_batch, file)
            common.upload_to_bucket(storage_client, result_location, path_file, BUCKET_RESULTS)
            result_locations.append(result_location)
    return result_locations

@Anna_Geller: The problem here is that parameters argument on the create_flow_run task expects a dictionary in the format {“param_name”: param_value} and you are currently passing just a string with parameter value.

This should fix it:

@task(max_retries=1, retry_delay=timedelta(seconds=300), state_handlers=[fail_handler], log_stdout=True)
def split_store_batches(metadatas: List[dict]) -> List[Dict[str, str]]:
    metadatas_batches = [{"metadatas_batch": batch_value} for batch_value in list(batch(metadatas, BATCH_SIZE_INDEX))]
    result_locations = []
    with tempfile.TemporaryDirectory() as tmp_dir:
        for metadatas_batch in metadatas_batches:
            today = prefect.context.get("today").split("-") # type: ignore
            file_id = uuid.uuid4()
            path_file = f"{tmp_dir}/{file_id}"
            result_location = f"{today[0]}/{today[1]}/{today[2]}/metadata_batches/{file_id}"
            with open(path_file, "w") as file:
                json.dump(metadatas_batch, file)
            common.upload_to_bucket(storage_client, result_location, path_file, BUCKET_RESULTS)
            result_locations.append({"metadatas_batch_location": result_location})
    return result_locations

@Jean-Baptiste_Six: Ah yes ! You already told me that, I just forgot after some manipulations ! Thx
It’s working
Should I use ‘wait_for_flow_run’ ? If yes, how to use it with the map ?
Smth like this ?

mapped_flow_run_ids = create_flow_run.map(
        flow_name=unmapped("index_flow"),
        project_name=unmapped("Document Pipeline"),
        parameters=result_locations
    )
wait_for_flow_run(mapped_flow_run_ids)

Or maybe I have to do :

wait_for_flow_run.map(mapped_flow_run_ids)

?

@Anna_Geller: Nice work! The latter should work and you can also pass extra arguments which are quite useful to see the logs in the UI:

wait_for_flow_run.map(flow_run_id=mapped_flow_run_ids, raise_final_state=unmapped(True), stream_logs=unmapped(True))

[anna_geller]

Another example:

Orchestrator-pattern

Child flow

 from prefect import Flow, Parameter, task


@task(log_stdout=True)
def hello_world(user_input: str):
    print(f"hello {user_input}!")


with Flow("dummy-child-flow") as flow:
    param = Parameter("user_input", default="world")
    hw = hello_world(param)

if __name__ == '__main__':
    flow.register("p")

Parent flow

 from prefect import Flow, unmapped, task
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
from prefect.executors import LocalDaskExecutor
from prefect.triggers import all_finished


with Flow("mapped_flows", executor=LocalDaskExecutor()) as flow:
    parameters = [
        dict(user_input="Arthur"),
        dict(user_input="Marvin"),
        dict(user_input="Ford"),
    ]
    mapped_flow_ids = create_flow_run.map(
        parameters=parameters,
        flow_name=unmapped("dummy-child-flow"),
        project_name=unmapped("p"),
    )
    wait_for_flow_run.map(mapped_flow_ids, raise_final_state=unmapped(True))


if __name__ == "__main__":
    flow.register("p")

---

Another example splitting the same flow into 8 flow runs grouped based on parameter values

 from prefect import Flow, task, unmapped
from prefect.tasks.prefect import create_flow_run
from prefect.executors import LocalDaskExecutor


@task
def generate_thousand_numbers(start, stop, step):
    nrs = range(start, stop, step)
    return list(nrs)


with Flow("mapped_flows", executor=LocalDaskExecutor()) as flow:
    flow_run_1 = generate_thousand_numbers(1, 1000, 1)
    flow_run_2 = generate_thousand_numbers(1000, 2000, 1)
    flow_run_3 = generate_thousand_numbers(2000, 3000, 1)
    flow_run_4 = generate_thousand_numbers(3000, 4000, 1)
    flow_run_5 = generate_thousand_numbers(4000, 5000, 1)
    # ... until 8
    parameters = [
        dict(list_of_numbers=flow_run_1),
        dict(list_of_numbers=flow_run_2),
        dict(list_of_numbers=flow_run_3),
        dict(list_of_numbers=flow_run_4),
        dict(list_of_numbers=flow_run_5),
        # ... until 8
    ]
    mapped_flows = create_flow_run.map(
        parameters=parameters,
        flow_name=unmapped("dummy-child-flow"),
        project_name=unmapped("community"),
    )