Channels
#flytekit
Title
# flytekit
j
Jay Ganbat
04/04/2022, 11:25 PMDoes
Mapped Task Copy code
E TypeError: Failed to convert return value for var o0 for function balrog_pipeline.workflows.primer_qc.process_expected_primer_qc_workflow.mapper_calculate_spike_in_metrics_in_fastq_dir_2 with error <class 'TypeError'>: No automatic conversion found from type <class 'flytekit.core.promise.Promise'> to FlyteFile.Supported (os.PathLike, str, Flytefile) Copy code
out_spike_in_metrics_list = map_task(
calculate_spike_in_metrics_in_fastq_dir,
metadata=TaskMetadata(retries=DEFAULT_DYNAMIC_TASK_RETRY),
)(task_params=task_params_list)calculate_spike_in_metrics_in_fastq_dir Copy code
@BASE_TASK_WORKER.run_task_with_args()
def calculate_spike_in_metrics_in_fastq_dir(
task_params: dict[str, FlyteFile]
) -> FlyteFile:y
Yee
04/04/2022, 11:48 PM
one thing, i think that should be
typing.Dictand what is base_task_worker?
j
Jay Ganbat
04/04/2022, 11:50 PMoh we are on python 3.9
so native typing works without specifically importing typing
y
Yee
04/04/2022, 11:50 PM
got it
j
Jay Ganbat
04/04/2022, 11:51 PMbase_task_worker is a decorator
that just adds resource request and cache versions to task/dynamic tasks
y
Yee
04/04/2022, 11:51 PM
and just to be clear, running the calculate_spike not in a map task works right?
j
Jay Ganbat
04/04/2022, 11:51 PMyup running remotely works
y
Yee
04/04/2022, 11:51 PM
if you try with just one element of the task_params_list
no i mean locally
j
Jay Ganbat
04/04/2022, 11:52 PMlocally doesnt work, getting this Promise error
y
Yee
04/04/2022, 11:52 PM
i mean just the
calculate_spike_in_metrics_in_fastq_dirit fails with the same error?
j
Jay Ganbat
04/04/2022, 11:52 PMoh that i havent tested but i imagine it works fine
since other regular flyt tasks run just fine
y
Yee
04/04/2022, 11:54 PM
let me try to repro this.
just look at the top file
does that kinda reflect what you were doing?
that works locally on my end
(the lp.py file is my attempt at debugging the other issue, ignore that for now)
j
Jay Ganbat
04/05/2022, 12:56 AMHi Yee I'll look into that in an hour currently not at my computer 😅
hmm yeah it is pretty much that.
Although input to the map task was generated by another python task, like this
Copy code
@workflow(failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def process_expected_primer_qc_workflow(
in_spec_path: FlyteFile,
) -> tuple[FlyteFile, FlyteFile, FlyteFile]:
task_params_list, can_run_contamination_metric = build_task_params(
in_spec_path=in_spec_path
)
out_spike_in_metrics_list = map_task(calculate_spike_in_metrics_in_fastq_dir, metadata=TaskMetadata(retries=DEFAULT_DYNAMIC_TASK_RETRY),
)(task_params=task_params_list)hmm ill try to replicate it on a sandbox to see if that would work
ok i think i know the issue
but still wierdly why your example is working
i have a sandbox workflow like this
Copy code
@task
def mappable_task(input: Dict[str, int]) -> Dict[str, int]:
return {"number": input["number"] ** 2, "is_even": input["is_even"]}
@task
def get_numbers(input: int) -> List[Dict[str, int]]:
return [
{"number": i, "is_even": 1 if i % 2 == 0 else 0} for i in range(input)
]
@workflow
def mappable_workflow(limit: int) -> List[Dict]:
mappable_inputs = get_numbers(input=limit)
results_mapped_task = map_task(mappable_task, metadata=TaskMetadata(retries=1))(
input=mappable_inputs
)
return results_mapped_task Copy code
raise AssertionError(f"Failed to convert value of output {k}, expected type {v}.") from e
AssertionError: Failed to convert value of output o0, expected type typing.List[typing.Dict].i think it specific to Map types, this example was a generic Dict type before and it was working so Generic Struct works but when i type it then it fails
@Yee do you think its because i am on py3.9 with no native typing annotation
y
Yee
04/06/2022, 4:38 PM
not sure… can you try with the typing.Dict?
or are you saying that doesn’t work
it should, we run our unit tests on 3.7-3.10
j
Jay Ganbat
04/06/2022, 4:39 PMnvm yeah its not workoing the example above is using the
from typing import Dict, Listhmm ok above example is working on v0.31.0 i was on 0.30.3 in the sandbox 🤦
sorry for the alarm 😅
8 Views