Hello We would like to get some help with an issue we are fa

Question

Hello We would like to get some help with an issue we are facing when we use Structured Dataset as an output of a task Intermittently the task gets stuck in Queued status Sample Code in thread

Govind Raghu · Answer

``` task def struct task forecast config FXConfig gt OutputSchema config df = forecast config open pd DataFrame all fcst start = config df fcst start 0 data = John 10 Alice 15 Bob 14 fcst start 3 df = pd DataFrame data columns= name age return OutputSchema dataframe=df uri=utils gen bq path model name def struct model standalone forecast config FXConfig gt OutputSchema return struct task forecast config=forecast config task def get assumption version id str gt FXConfig d = fcst start 2022 09 10 fcst end 2022 09 10 version version id df = pd DataFrame data=d return StructuredDataset dataframe=df workflow def struct model onemodel version id str gt OutputSchema forecast config = get assumption version id=version id return struct model standalone forecast config=forecast config ```

Govind Raghu · Answer

Schemas ```OutputSchema = Annotated StructuredDataset kwtypes name=str age=int FXConfig = Annotated StructuredDataset kwtypes fcst start=str fcst end=str version=str ```

Govind Raghu · Answer

After the timeout we see an error similar to below show up in the console ```Failed to check Catalog for previous results unexpected artifactData o0 type structured dataset type lt columns lt name forecast version literal type lt simple STRING gt gt columns lt name fromCurrency literal type lt simple STRING gt gt columns lt name toCurrency literal type lt simple STRING gt gt columns lt name rate literal type lt simple FLOAT gt gt columns lt name year start rate literal type lt simple FLOAT gt gt gt does not match any task output type structured dataset type lt columns lt name forecast version literal type lt simple STRING gt gt columns lt name fromCurrency literal type lt simple STRING gt gt columns lt name toCurrency literal type lt simple STRING gt gt columns lt name rate literal type lt simple FLOAT gt gt columns lt name year start rate literal type lt simple FLOAT gt gt format parquet gt ```

Ketan (kumare3) · Answer

Cc < Yee>

Ketan (kumare3) · Answer

< Kevin Su>

Kevin Su · Answer

< Govind Raghu> Which flytekit version are you using

Govind Raghu · Answer

flytekit version is 1 1 1

Alireza · Answer

< Govind Raghu> Can you please add ` task` above `def struct model standalone` functions Your error says ``` type structured dataset type does not match any task output type ``` My initial guess is that this is expecting some output from a task So adding ` task` might help Also can you please run `df dtypes` on your dataset and share that output I would like to see if the dataset has a consistent datatype on all rows of every column I am also thinking about the ` gt format parquet gt `` in the error message but we can dig in that after your initial tests

Dylan Wilder · Answer

we ve tried this < Alireza> but this failure is at execution time not registration and we have this pattern elsewhere and it appears to be directly related to structured datasets as flyteschemas work fine

Dylan Wilder · Answer

`struct model standalone` returns an actual flyte task we use this pattern to share common workflow dags without the runtime overhead of actually using a subworkflow

Alireza · Answer

< Dylan Wilder> have you tried to delete the `struct model standalone` i e to use the ``` workflow def struct model onemodel version id str gt OutputSchema forecast config = get assumption version id=version id return struct task forecast config=forecast config ``` If yes did you see the same error

Govind Raghu · Answer

yes I tried after removing standalone method too but it still resulted in the same issue

Alireza · Answer

I see Thank you for the quick check In our doc the template looks like this <https docs flyte org projects cookbook en latest auto core type system structured dataset html structured dataset> ``` task def get subset df df Annotated StructuredDataset subset cols gt Annotated StructuredDataset subset cols df = df open pd DataFrame all df = pd concat df pd DataFrame 30 columns= Age On specifying BigQuery uri for StructuredDataset flytekit writes a pandas dataframe to a BigQuery table return StructuredDataset dataframe=df ``` I am still trying to see the possible differences Can you please reorganize your first task to look like the above and see if getting same error

Kevin Su · Answer

< Dylan Wilder> I ran your example in flytekit but that s working for me Do you have any other workflows that can produce the same error The root cause is that format is different between input type and expected type ``` structured dataset type lt columns lt name forecast version literal type lt simple STRING gt gt columns lt name fromCurrency literal type lt simple STRING gt gt columns lt name toCurrency literal type lt simple STRING gt gt columns lt name rate literal type lt simple FLOAT gt gt columns lt name year start rate literal type lt simple FLOAT gt gt gt does not match any task output type structured dataset type lt columns lt name forecast version literal type lt simple STRING gt gt columns lt name fromCurrency literal type lt simple STRING gt gt columns lt name toCurrency literal type lt simple STRING gt gt columns lt name rate literal type lt simple FLOAT gt gt columns lt name year start rate literal type lt simple FLOAT gt gt format parquet gt ```

Kevin Su · Answer

Here is the workflow I executed ```from typing import Annotated import pandas as pd from flytekit import task workflow StructuredDataset kwtypes OutputSchema = Annotated StructuredDataset kwtypes name=str age=int FXConfig = Annotated StructuredDataset kwtypes fcst start=str fcst end=str version=str def struct task forecast config FXConfig gt OutputSchema config df = forecast config open pd DataFrame all fcst start = config df fcst start 0 data = John 10 Alice 15 Bob 14 fcst start 3 df = pd DataFrame data columns= name age return OutputSchema dataframe=df uri= <bq flyte test 340607 dataset test1> task def struct model standalone forecast config FXConfig gt OutputSchema return struct task forecast config=forecast config task def get assumption version id str gt FXConfig d = fcst start 2022 09 10 fcst end 2022 09 10 version version id df = pd DataFrame data=d return StructuredDataset dataframe=df workflow def struct model onemodel version id str gt OutputSchema forecast config = get assumption version id=version id return struct model standalone forecast config=forecast config if name == main struct model onemodel version id= id ```

Govind Raghu · Answer

yes it works most of the time this issue is intermittent usually happens when multiple workflows are running together

Govind Raghu · Answer

here s another example for which I was able to reproduce this issue ```Schema1 = Annotated StructuredDataset kwtypes name=str age=int Schema2 = Annotated StructuredDataset kwtypes len=int task def t1 gt Schema1 df = pd DataFrame name dylan steve age 33 32 return StructuredDataset df task def t2 sd Schema1 gt Schema2 df = pd DataFrame len len sd open pd DataFrame all table id = str uuid uuid4 bq uri = f <bq sp one model one> model dev table id return StructuredDataset df uri=bq uri workflow def wf gt Schema2 return t2 sd=t1 ```

Kevin Su · Answer

nice taking a look

Govind Raghu · Answer

this works fine on single execution but if I try to run 2 or 3 executions simultaneously I run into the issue

Dylan Wilder · Answer

looks like it s specific to writing to bq

Dylan Wilder · Answer

or reading from

Kevin Su · Answer

< Dylan Wilder> Did you use caching I can reproduce it if cache enabled

Govind Raghu · Answer

Yes caching is enabled on the task

Kevin Su · Answer

yes there is a bug in flytekit will create a pr shortly To workaround this disable cache in flytekit BQ can cache result as well

Govind Raghu · Answer

Thanks Kevin

Dylan Wilder · Answer

wait the above examples do not have caching though

Govind Raghu · Answer

< Dylan Wilder> the examples I tested were using the custom decorator that enabled caching by default

Dylan Wilder · Answer

oh good point smile