Hi folks!, I'm a disgruntled airflow user here looking to learn more about prefect. I've started to port over a simple ETL workflow with some success.

Hi @Bob Colner and welcome! Anytime you call a task using the functional API, a new task is created. That means that in the following example,

a

and

b

are completely independent tasks:

with Flow("example") as flow:
    a = my_task()
    b = my_task()

thanks, I'm my case there is no output to store. Each task is updating tables in a remote DB

Sorry I didn’t mean the output of the task itself but rather the return value of calling the task; (see my code snippet)

oh, so in the flow context the output is special?

the result of calling a task instance is a new copy of the task in the same way that

f() and f()

are two totally different function calls with distinct return values; this tutorial actually covers the distinction: https://docs.prefect.io/core/advanced_tutorials/task-guide.html#adding-tasks-to-flows See the section header “Calling Tasks creates copies”

I'll check that out, still not understanding. In my example each task is unique and only needs to be run once. but the ordering matters

So it’s all about what API calls you make to build your flow; here’s an example of what I mean:

@task
def my_task():
    return 42

with Flow("example") as flow:

   # calling my_task produces a copy
   # that is distinct from my_task
   # but has the same runtime logic

    a = my_task()
    b = my_task(upstream_tasks=[a])

which is a flow with two tasks as can be seen in the following DAG viz:

thanks, and I've got that basic flow working if you . see my DAG viz.

no worries at all! I understand where you’re coming from. Give me a few minutes and I’ll refactor your code snippet to work as you expect

thanks so much

Anytime! Try this out and see if it does what you expect:

with Flow("core-etl", schedule=schedule) as flow:

    sub_task = dims_subscribers(
        upstream_tasks=[events_subscriptions]
    )

    made_task = dims_devices_made(
        upstream_tasks=[dims_devices_made_installed, dims_adjust]
    )

    inst_task = dims_devices_instasize(
        upstream_tasks=[dims_devices_instasize_installed, dims_adjust]
    )

    dims_merged_subscribers(
        upstream_tasks=[inst_task, sub_task]
    )

nice! ah much better

yup exactly!

thanks again, I'm liking the design of the tool/service -reminds me a bit of AWS SWS

anytime! and thanks - let us know if you have any other questions