https://flyte.org logo
Powered by
#ask-the-community
Title
# ask-the-community
v

Visak

03/21/2023, 11:01 PM
Hi all, a flyte-newbie here , I want to write a workflow that runs the same task sequentially in a workflow. Any thoughts/pointers to examples as to how I can do this in Flyte? For example, here is a dummy case where I want to run the task ‘add_1’ sequentially, currently flute would run these simultaneously As there is no dependency between 2 calls
Copy code
@task
Def add_1(n : float) -> float:
	return n + 1

@workflow
Def some_name():
	# I want these two steps to run sequentially instead of parallel
	c1 = add_1(1.0)
	c2 = add_1(2.0)
j

Jay Ganbat

03/21/2023, 11:06 PM
chaining tasks might be more appropriate here
v

Visak

03/21/2023, 11:13 PM
Thanks! ill look into these. An additional question that comes to my mind is if chaining would be possible in a loop. For example, how could I create dependency or chain in this case :
Copy code
@workflow
Def some_name():
	# I want these two steps to run sequentially instead of parallel
	for item in list:
        add_1(item)
j

Jay Ganbat

03/21/2023, 11:17 PM
Interesting idea, it might not be possible to use list inside workflow, does 
list
has a predefined values or will it be determined at runtime? because workflow is evaluated during registration. You might be able to do that in dynamic task @Ketan (kumare3) any insight 😅
v

Visak

03/21/2023, 11:23 PM
The list can be predefined this case. Im more curious about if we can introduce chaining/dependency on subsequent calls of the same 
task
in a loop
j

Jay Ganbat

03/21/2023, 11:26 PM
i wonder if this works 😆
Copy code
result = SOME VALUE
for item in list:
    result = add_1(item, result)
never though about this before
v

Visak

03/21/2023, 11:30 PM
😅 can give it a shot! my hunch is that since flyte is an acyclic graph, this closed loop dependency will be not possible? unless there is some way to wait/sleep until a task finishes
v

varsha Parthasarathy

03/22/2023, 12:08 AM
Does wait_for_input(..) help here @Jay Ganbat or @Ketan (kumare3)
I haven’t used this construct before, but very curious to understand 😄
k

Ketan (kumare3)

03/22/2023, 1:21 AM
@Visak / @Jay Ganbat as long as the list is predefined chaining should work - just need to create node or use promise chaning - capture it in a local Variable. I will try to write an example later If you want list to be dynamic, then you have to use dynamic workflows
Wait for input does not provide inputs on which you can iterate in a workflow - you only get a promise. You can materialize the promise in a task and then run it in dynamic
v

Visak

03/22/2023, 1:26 AM
Thanks! Yes, an example would be super helpful here !
v

varsha Parthasarathy

03/22/2023, 2:16 AM
Copy code
x: will be different for every run 

for i in range(x):
        splits = split_inputs(inputs=input_tasks,s=i)
        r = flytekit.map_task(
            run_maptask_fpo,
            concurrency=50,
            min_success_ratio=0.9)(
            curation_input=splits)
        result.append(r)
@Ketan (kumare3) - I want flytekit.map_task(..) to be launched once, wait for r , append to result and again call flytekit.map_task(..) for x times. But currently all map_tasks are launched x times, all at once.
k

Ketan (kumare3)

03/22/2023, 4:46 AM
hmm then the code should look different right?
that is implicit dependency
Crazily this works!!!
Copy code
from flytekit import task, workflow, map_task, dynamic
import typing

@task
def square(i: int) -> int:
    return i*i

@task
def splits(i: int) -> typing.List[int]:
   return list(range(i))

@task
def join(new_v: typing.List[int], v: typing.List[int]) -> typing.List[int]:
   v.extend(new_v)
   return v

@dynamic
def do(i: int, x: int, v: typing.List[int]) -> typing.List[int]:
    s = splits(i=i)
    new_v = map_task(square)(i=s)
    v = join(new_v=new_v, v=v)
    if i == x - 1:
       return v
    return do(i=i+1, x=x, v=v)

@workflow
def wf(x: int):
    do(i=1, x=x, v=[])
cc @varsha Parthasarathy
cc @Niels Bantilan / @Eduardo Apolinario (eapolinario) / @Yee And interesting i got it right in one go - in vim 😄. hmmm, functional programming @varsha Parthasarathy at this point what you need is async workflows. Something that we are working / specing now
Screenshot 2023-03-21 at 10.09.52 PM.png
Screenshot 2023-03-21 at 10.10.11 PM.png
j

Jay Ganbat

03/22/2023, 5:11 AM
Whoa that is crazy but awesome 🤣
v

Visak

03/22/2023, 5:39 AM
This is pretty dope! thanks
k

Ketan (kumare3)

03/22/2023, 5:53 AM
Haha, but not very performant
Also cc @James Sutton and @Evan Sadler
2 Views
Powered by