Also where can i get some info on how the results are gettin Flyte #ask-the-community

Also where can i get some info on how the results ...

Jay Ganbat

08/01/2023, 5:30 PM

Also where can i get some info on how the results are getting stored in datacatalog, i kept getting this error

Copy code

errorsyncing'my-development/azcrpvdss4xngltqgpd6': FailedtocheckCatalogforpreviousresults: unexpectedartifactData: [trimmed_fastqs]
type: 
[collection_type: <map_value_type: <union_type: <variants: <blob: <>>variants: <union_type: <variants: <blob: <>structure: <tag: "FlyteFilePath">>>>>>>]
does not match any task output type: 
[collection_type: <map_value_type: <union_type: <variants: <blob: <>structure: <tag: "FlyteFilePath">>variants: <simple: NONEstructure: <tag: "none">>>>>]

Kevin Su

08/01/2023, 6:13 PM

seems like a bug in the type checker. could you share the task code? I can take a look into it

Dan Rammer (hamersaw)

08/01/2023, 6:21 PM

@Jay Ganbat can you share a minimal repro of this? I think it's actually a part of a greater issue we're debugging right now.

Jay Ganbat

08/01/2023, 6:56 PM

yeah the most annoying thing is i just cannot replicate it in a sandbox, but here is the task signature

Copy code

@BASE_TASK_WORKER.run_dynamic_task_with_args()
def process_chunked_fastqs_to_raw_bams_task(
    in_sample_name: str,
    in_spec_path: FlyteFile,
    in_fastq_pairs: list[dict[str, Optional[FlyteFile]]],
    in_expected_spikein: str,
    out_gcs_dir: str,
) -> NamedTuple(  # type: ignore
    "process_chunked_fastqs_to_raw_bams_outputs",  # noqa
    trimmed_fastqs=list[dict[str, Optional[FlyteFile]]],
    ipc_fastqs=list[dict[str, Optional[FlyteFile]]],
    raw_bams_to_merge=list[FlyteFile],
    raw_adapter_trim_metrics=list[Optional[FlyteFile]],
    raw_umi_metrics=list[dict[str, Optional[FlyteFile]]],
    spike_in_metrics=list[Optional[FlyteFile]],
):

and the error comes from the output

Jay Ganbat

08/01/2023, 6:58 PM

so above task is executed as part of a large workflow and that workflow crashes whenever i try to relaunch or recover. if things wont work, maybe i should just do

list[dict[str, FlyteFile]]

and deal with the empty results in the concrete tasks

Jay Ganbat

08/01/2023, 7:58 PM

@Dan Rammer (hamersaw) do you have a github issue with more info regarding your big issue 😅

Jay Ganbat

08/01/2023, 10:25 PM

oh i was jjust reading through the release note, noticed this

Copy code

Union | optional return types supported by @kumare3 in #1703

it is on

v1.7.1b1

and we are on flytekit 1.4.2 maybe this was the issue 🤔 im upgrading our flytekit right now

Dan Rammer (hamersaw)

08/02/2023, 10:03 PM

@Jay Ganbat here is the github issue. I have a fix coming shortly - it looks to be the exact same problem you're facing, namely the double

union

Jay Ganbat

08/02/2023, 10:24 PM

thank you very much, yeah the issue is exactly the same. maybe the way it was triggered is different, in our case its triggered on relaunch/recover operation. i will look forward to the fix and test it out.

4 Views

Open in Slack

Previous Next