Hi all does Recover on dynamic task work properly we have a

Question

Hi all does Recover on dynamic task work properly we have a dynamic task that returns a Dict with following `sample name sample workflow results` When some sample fails and we recover unfortunately the output mapped dict is all shuffled up and no longer matches with the sample name key For example we expect the certain dynamic task returns `filename gt filename suffix` map like below ``` counter 1 1 counter 2 2 counter 3 3 counter 4 4 counter 5 5 counter 6 6 ``` and i forcefully fail some tasks and some successfully completes Then when i recover i got this instead ``` counter 1 5 counter 2 6 counter 3 3 counter 4 4 counter 5 5 counter 6 4 ``` example dynamic task is below ``` dynamic cache=True cache version=get workflow version def compute average for chunks chunks Dict str FlyteFile infile FlyteFile gt Dict str int suffix dict = for chunk name chunk path in chunks items infile suffix = get file suffix task chunk path=chunk path infile=infile suffix dict chunk name = infile suffix return suffix dict```

Jay Ganbat · Answer

I was hoping to get a deterministic output here sweat smile would love to have someone replicate this issue `infile`input has a file in cloud that we read to forcefully fail some tasks in `get file suffix task`

Ketan (kumare3) · Answer

this looks wrong

Jay Ganbat · Answer

Interestingly if we expand it to the underlying tasks they are all correct I think it s when we convert from python literal to flyte specific job that is causing this thinking face

Ketan (kumare3) · Answer

hmm sounds like it

Ketan (kumare3) · Answer

is this map task actually no this is dynamic

Ketan (kumare3) · Answer

cc < Dan Rammer hamersaw > < Haytham Abuelfutuh> this looks odd and dangerous

Jay Ganbat · Answer

here is our cluster setup ``` name newTag v1 0 1 sha256 b41b322e7a926f7e8efd351e6e51e246452477e28659de0c6152e2a11a7b7e36 name newTag v1 1 46 sha256 2359a58d756186992f7eba56743874aa8180fe8d71a7c3e1436141f59f117b31 name newTag v1 2 5 sha256 164d946afc39762c61b343b69255414deca12d2d0b14d2dbfb82fc0ac31a1da3 name newTag v1 1 40 sha256 db51f3ba890c4198a02f4299e68ec3177a5a42f5dee88bea6691d9db6e906074``` and we are on Flytekit `1 1 1` not sure if this issue is gone in the latest versions

Dan Rammer (hamersaw) · Answer

gt Interestingly if we expand it to the underlying tasks they are all correct I think it s when we convert from python literal to flyte specific job that is causing this thinking face < Jay Ganbat> can you expand on this I ve done a bunch of testing locally and am unable to repro When you check the output of the dynamic in the recovered workflow it is correct but some downstream node is incorrect

Jay Ganbat · Answer

output of dynamic task after recovering is all wrong in the final output but if you expand the individual tasks they have the correct output

Dan Rammer (hamersaw) · Answer

interesting since the dynamic task is cached in this case if you just rerun the workflow rather than recovering it is the cached data from the dynamic task output read as expected Or is this incorrect as well

Dan Rammer (hamersaw) · Answer

oh wow i actually think i know exactly what is happening here now that i dive into it more

Dan Rammer (hamersaw) · Answer

so using a Dict as input means that all of the data is unordered so on the initial run the mapping of `chunk name` to Flyte node names could be something like ``` counter 1 n0 counter 2 n1 counter 3 n2 counter 4 n3 ``` if this runs and you manually fail nodes `n2` and `n3` then these nodes will not be ran during the recovery rather their outputs will be reused however in the recovery we can not guarantee that the Dict is iterated in the same order so the recovery may now map `chunk name` to Flyte node names like ``` counter 3 n0 counter 2 n1 counter 4 n2 counter 1 n3 ``` In this case we see the outputs of `n0` and `n1` which previously succeeded are now used for counter 3 and counter 2 rather than the counter 1 and counter 2 that they should have

Jay Ganbat · Answer

yeah i think so if i relaunch everything works just fine

Jay Ganbat · Answer

its just this part ``` for chunk name chunk path in chunks items infile suffix average = compute average for chunk chunk path=chunk path infile=infile ``` is somehow shuffled Do you think it would work if we do not attempt to cache dynamic task thinking face

Jay Ganbat · Answer

nope disabling cache on dynamic task didnt change the result

Dan Rammer (hamersaw) · Answer

yeah sorry i would not expect it to i m running a few tests here locally to make sure i understand this correctly if so i think this is a bug that we ll have to address with a code change

Dan Rammer (hamersaw) · Answer

< Jay Ganbat> so it s as I feared I was able to repro locally now recovery over dynamic workflows may not work in every scenario because the Flyte node id s are not deterministic because the iteration order of python is not deterministic in your scenario above if you sorted the dict on key then it would be but a general fix will require Flyte code changes I will submit an issue shortly

Dan Rammer (hamersaw) · Answer

cc < Ketan kumare3 > < Haytham Abuelfutuh> < Yee> ^^^ recovery of dynamic workflows is broken because the Flyte node IDs are not deterministic when iterating over a python Dict We currently attempt to recover any subnodes that succeeded in a dynamic task but since the Flyte node IDs may not match between runs we can assign incorrect output In this scenario the error was incorrect output values but it could easily be misformatted as well ex n0 outputs int and n1 outputs string and they are swapped

Jay Ganbat · Answer

Thank you very much for looking into it So in the meantime after all results returned i could sort them before returning it in the dynamic task right

Dan Rammer (hamersaw) · Answer

I think you would have to sort before starting the iteration so something like ``` dynamic cache=True cache version=get workflow version def compute average for chunks chunks Dict str FlyteFile infile FlyteFile gt Dict str int suffix dict = for chunk name chunk path in sorted chunk items key = lambda ele ele 0 infile suffix = get file suffix task chunk path=chunk path infile=infile suffix dict chunk name = infile suffix return suffix dict``` This way Flyte will assign deterministic node IDs ex counter 1 will always be n0 counter 2 will always be n1 etc

Jay Ganbat · Answer

ohhh i see so its actually the input dict that is shuffled even if in flyte console it looks in order

Dan Rammer (hamersaw) · Answer

exactly if you iterate over a dict in python 10 times you could get 10 different orderings

Yee · Answer

not following what s happening here can you elaborate dan it s true that python map iteration order is not guaranteed but i don t follow why that matters what is happening in the recovery scenario

Dan Rammer (hamersaw) · Answer

so Flyte is saying in the previous run element 0 succeeded successfully lets just reuse that result but the input value for element 0 may be different in the second run

Jay Ganbat · Answer

oh wait yeah as long as key value pair is consistent it shouldnt matter right

Dan Rammer (hamersaw) · Answer

< Yee> hop on a call quick Jay feel free to join

Jay Ganbat · Answer

sure smile

Dan Rammer (hamersaw) · Answer

To join the video meeting click this link <https meet google com xnt qfyn yjh>

Jay Ganbat · Answer

can you share the issue here if you created one sweat smile

Dan Rammer (hamersaw) · Answer

< Jay Ganbat> please feel free to add more context if you think it will help We will try to get a fix out soon

Jay Ganbat · Answer

Hi < Dan Rammer hamersaw > do you think we would have similar issue with Map task too does inputs to map task preserved So we have a map task like this and i cant sort in puts since they will be `Promise` type during the hand over from map task input to run map task ``` in a dynamic task we have the following map task input = build map task inputs this should return a list of inputs run map task map task input ```

Dan Rammer (hamersaw) · Answer

Hey < Jay Ganbat> there shouldn t be any issue with map tasks Right now map tasks will not recover completed subtasks on the workflow recovery Caching should still work so if the first one succeeded and wrote to cache then when recovering the workflow it will read results from the cache

Dan Rammer (hamersaw) · Answer

This is one of a few issues with the current implementation of map tasks We have a proposal to introduce a new node type in the Flyte DAG namely ArrayNode which would remedy this issue Additionally it would allow for map tasks over different types ie Spark Ray etc

Jay Ganbat · Answer

Oh i meant the order of inputs would still cause an issue for map task

Dan Rammer (hamersaw) · Answer

< Jay Ganbat> yeah so I stand corrected in my answer based on testing for So the order of inputs will only matter if the map task was successful For example if the inputs for the first one are ` 0 1 2 3 4 ` ordered and all but `4` is successful currently map task subtask successes can not be recovered So even though ` 0 1 2 3 ` all completed successfully they will all be recomputed when recovering However if all of the subtasks succeeded and the recovery has an order of ` 4 3 2 1 0 ` then we will run into the same issue with outputs being set incorrectly Does this make sense

Jay Ganbat · Answer

Oh I see yeah that makes sense