Scott Asher
10/26/2020, 3:47 AMPrefect’s imperative API allows more fine-grained control. Its main advantage over the functional API is that it allows tasks to be set as upstream or downstream dependencies without passing their results. This allows you to create a strict ordering of tasks through state dependencies without also creating data dependencies.And then you give an example where you force a “PlusOne” task to run after a task that simply prints something. My question is - what if you WANT a data dependency? Can you do this through the imperative API? More code blocks to follow in thread so I don’t clutter up the main thread.
from prefect import Task, Flow
class RunMeFirst(Task):
def run(self):
print("I'm running first!")
class PlusOneTask(Task):
def run(self, x):
return x + 1
flow = Flow('My Imperative Flow')
plus_one = PlusOneTask()
flow.set_dependencies(
task=plus_one,
upstream_tasks=[RunMeFirst()],
keyword_tasks=dict(x=10))
This is at: https://docs.prefect.io/core/concepts/flows.html#overview
But what if I had:
from prefect import Task, Flow
class RandomNumber(Task):
def run(self):
return random.randint(0, 100)
class PlusOneTask(Task):
def run(self, x):
return x + 1
And I wanted to make the input from a RandomNumber
run the input to PlusOneTask
using the imperative api?flow = Flow('My Imperative Flow')
plus_one = PlusOneTask()
flow.set_dependencies(
task=plus_one,
keyword_tasks=dict(x=RandomNumber())
?Zanie
keyword_tasks: Mapping[str, object]: The results of these tasks will be provided to this task under the specified keyword arguments.
Scott Asher
10/26/2020, 3:18 PM