https://flyte.org logo
#ask-ai
Title
# ask-ai
m

Maciej Kopczyński

02/06/2024, 9:08 PM
@Glime I try to execute the same task multiple times (in parallel) on a list of inputs, so I am using
map_task
. The input for the map task is dynamically obtained (via database query), therefore I encapsulate that map task in a dynamic workflow. Below is a generalized example of what I am trying to do:
Copy code
@task
def get_dynamic_params(...) -> list[str]:
    # Task implementation
    pass


@task
def my_task(dynamic_param: str, param1: str, param2: str,) -> DataFrame:
    # Task implementation
    pass


@dynamic
def my_dynamic_workflow(
    dynamic_params: list[str],
    param1: str,
    param2: str,
) -> list[DataFrame]:
    partial_my_task = functools.partial(
        my_task,
        param1: str,
        param2: str,
    )
    return map_task(partial_my_task)(dynamic_param=dynamic_params)

@workflow
def my_static_workflow(
    param1: str,
    param2: str,
) -> list[...]:
    dynamic_params = get_dynamic_params()
    my_data = my_dynamic_workflow(
        dynamic_params=dynamic_params,
	param1 = param1,
	param2 = param2,
    )
    # Workflow implementation
    pass
When trying to register the workflow, I am getting following errors:
Copy code
Error 0: Code: MismatchingTypes, Node Id: end-node, Description: Variable [o0] (type [collection_type:<union_type:<variants:<map_value_type:<structured_dataset_type:<> > structure:<tag:"Typed Dict" > > variants:<simple:NONE structure:<tag:"none" > > > > ]) doesn't match expected type [collection_type:<map_value_type:<structured_dataset_type:<> > > ].
	Error 1: Code: ParameterNotBound, Node Id: end-node, Description: Parameter not bound [o0].
g

Glime

02/06/2024, 9:09 PM
Based on the context provided and the error messages you're encountering, it seems like there are a couple of issues with how the workflow and dynamic workflow are set up, particularly around the use of
map_task
within a dynamic workflow and the handling of its outputs. 1. ​*Mismatching Types Error*​: This error suggests that there's a discrepancy between the expected output type of your dynamic workflow and what it's actually returning. The error message indicates that the system expects a collection of
DataFrame
objects but is getting something else. This could be due to how the
map_task
outputs are being handled or returned in the dynamic workflow. Ensure that the
map_task
within the dynamic workflow is correctly set up to return a list of
DataFrame
objects as expected by the workflow's output. 2. ​*Parameter Not Bound Error*​: This error typically occurs when a workflow or task expects an input parameter that hasn't been provided or is incorrectly referenced. In the context of your dynamic workflow, it's possible that the way outputs are being returned from the
map_task
or how the dynamic workflow's outputs are defined might be causing this issue. Ensure that all expected outputs of the dynamic workflow are correctly defined and that the `map_task`'s outputs are being correctly captured and returned. Given the complexity of using
map_task
within a dynamic workflow, especially with dynamic parameters, here are a few suggestions: • ​*Review
map_task
Usage*​: Ensure that the
map_task
is correctly set up to handle a list of inputs and that its usage within a dynamic workflow aligns with Flyte's expectations. The
map_task
should be capable of processing each item in the list of dynamic parameters independently and in parallel. • ​*Check Dynamic Workflow Outputs*​: Verify that the dynamic workflow's outputs are correctly defined and match what the
map_task
returns. The dynamic workflow should return a list of
DataFrame
objects if that's what the subsequent steps or the calling workflow expects. • ​*Partial Function with `map_task`*​: When using
functools.partial
with
map_task
, ensure that the partial function is correctly set up and that all non-dynamic parameters are correctly passed. This setup can sometimes lead to confusion or errors if not done correctly. Without the ability to directly modify your code or see the full implementation, these suggestions are based on common issues and best practices. If these suggestions do not resolve your issue, consider providing more detailed code snippets or error logs for further analysis. --- This is an AI-generated response, take its recommendations with caution. Please help us improve the model accuracy by giving this response a thumbs up/thumbs down reaction. 👍 button 👎 button