I m also getting this error when trying to execute a workflo Flyte #ask-the-community

I'm also getting this error when trying to execute...

Fhuad Balogun

03/16/2023, 3:13 PM

I'm also getting this error when trying to execute a workflow. I've checked the types, they seem to be consistent.

Copy code

TypeTransformerFailedError: Type of Val 'typing.Dict' is not an instance of typing.Dict[str, str]

Evan Sadler

03/16/2023, 3:21 PM

Hello! I am from the community and happy to help debug. If you can paste a minimal working example that reproduces the error, I can work through it with you.

Fhuad Balogun

03/16/2023, 3:57 PM

Thank you @Evan Sadler

Copy code

class FlyteRemote:
    def __init__(self):
        ...

    def _register_execute_workflow(
        self,
        wf: flytekit.core.workflow.WorkflowBase,
        inputs: Dict,
    ) -> flytekit.remote.executions.FlyteWorkflowExecution:

        self.remote.register_workflow(wf, self.serialization_settings)

        fetched_wf = self.remote.fetch_workflow(
            name=wf.name,
            version=self.version,
        )

        execution_response = self.remote.execute(
            entity=fetched_wf,
            inputs=inputs,
            version=self.version,
            image_config=self.image_config,
        )

        return execution_response

    def run_regular_simulation_wf(
            self,
            inputs: Dict,
            ) -> flytekit.remote.executions.FlyteWorkflowExecution:

        wf_response = self._register_execute_workflow(wf=regular_simulation_cpu_wf, inputs=inputs)

Copy code

@workflow
def regular_simulation_cpu_wf(
    param1: Challenge,
    param2: List[int],
    param3: str,
    param4: str,
    param5: module.Model,
    param6: module.Explorer,
    param7: module.Landscape,
    param8: int,
    param9: str,
):

    return simulation_cpu_task(
        param1=param1,
        param2=param2,
        param3=param3,
        param4=param4,
        param5=param5,
        param6=param6,
        param7=param7,
        param8=param8,
        param9=param9,
    )

Evan Sadler

03/16/2023, 4:17 PM

This looks like a type issue with the Flyte workflow and task. I am noticing there is no typing for the output of the workflow, but I need to see •

simulation_cpu_task

to be able to judge more

Evan Sadler

03/16/2023, 4:22 PM

Also using

pyflyte run

from the command line might help isolate if the issues faster. Because you can nest workflows, you can create one for testing that wraps

regular_simulation_cpu_wf

and pass in defaults. Alternatively you can call

regular_simulation_cpu_wf

from inside a notebook too like normal python.

Copy code

@workflow
def test_regular_simulation_cpu_wf():
   regular_simulation_cpu_wf([pass in values])

Fhuad Balogun

03/16/2023, 4:42 PM

@Evan Sadler the

regular_simulation_cpu_wf

is a one-task workflow, so the parameters of that workflow is the same as the parameters of the task. Below is the snippet of the

simulation_cpu_task

Copy code

@task(requests=Resources(cpu="1", mem="1Gi"))
def simulation_cpu_task(
    param1: Challenge,
    param2: List[int],
    param3: str,
    param4: str,
    param5: module.Model,
    param6: module.Explorer,
    param7: module.Landscape,
    param8: int,
    param9: str,
) -> List[Campaign]:
    """CPU simulation task"""
    simulation = module.Simulation(
        param1=param1,
        param2=param2,
        param3=param3,
        param4=param4,
        param5=param5,
        param6=param6,
        param7=param7,
        param8=param8,
        param9=param9,
    )
    experiments = simulation.run_sim()
    return experiments

Thank you very much. I'm currently testing out your suggestions to see if they work. I will revert.

152 Views

Open in Slack

Previous Next