Joe Hartshorn
09/28/2023, 12:08 PMmap_task
and solved it so I thought I’d share the solution in case anyone else does too.
I was using tasked wrapped in partial
and the curried argument was a FlyteFile
. Getting an error like
TypeError: object of type 'FlyteFile' has no len()
Here’s an simplified example (task bodies omitted for brevity
@task()
def download_files() -> list[FlyteFile]:
...
return file_list
@task()
def get_checkpoint() -> FlyteFile:
...
return checkpoint_file
@task()
def process_file(file: FlyteFile, model: FlyteFile) -> dict:
...
return result
@workflow
def process_all_files() -> list[dict]:
file_list = download_files()
model = get_checkpoint()
process_using_model = partial(
process_file,
model=model,
)
result = map_task(process_using_model)(file=file_list)
return result
The error was in the task definition, def process_file(model: FlyteFile, file: FlyteFile)
should have been process_file(file: FlyteFile, model: FlyteFile)
. Notice how the arguments for the map task are defined after the argument which is supplied from the list.
This was a tough one to work out and I think should either be changed to make it more generic, or mentioned in the docs.Ketan (kumare3)
Niels Bantilan
09/28/2023, 4:22 PMSlackbot
09/28/2023, 4:22 PMNiels Bantilan
09/29/2023, 6:56 PMJoe Hartshorn
10/03/2023, 7:57 AM