13

My actual workloads that should be run as tasks within a Prefect flow are all packaged as docker images. So a flow is basically just "run this container, then run that container".

However, I'm unable to find any examples of how I can easily start a docker container as task. Basically, I just need docker run from a flow.

I'm aware of https://docs.prefect.io/api/latest/tasks/docker.html and tried various combinations of CreateContainer and StartContainer, but without any luck.

theDmi
  • 17,546
  • 6
  • 71
  • 138

1 Answers1

13

Using the Docker tasks from Prefect's Task Library could look something like this for your use case:

from prefect import task, Flow
from prefect.tasks.docker import (
    CreateContainer,
    StartContainer,
    GetContainerLogs,
    WaitOnContainer,
)

create = CreateContainer(image_name="prefecthq/prefect", command="echo 12345")
start = StartContainer()
wait = WaitOnContainer()
logs = GetContainerLogs()


@task
def see_output(out):
    print(out)


with Flow("docker-flow") as flow:
    container_id = create()
    s = start(container_id=container_id)
    w = wait(container_id=container_id)

    l = logs(container_id=container_id)
    l.set_upstream(w)

    see_output(l)

flow.run()

This snippet above will create a container, start it, wait for completion, retrieve logs, and then print the output of echo 12345 to the command line.

Alternatively you could also use the Docker Python client directly in your own tasks https://docker-py.readthedocs.io/en/stable/api.html#module-docker.api.container

Josh
  • 293
  • 1
  • 5
  • This works, however it creates for tasks just to run one docker container. This makes large flows hard to manage. I guess I take the second route you suggested, like this I can create a single "docker run" task. – theDmi Jun 23 '20 at 09:10
  • Hi @theDmi - you could merge them into one task via ```class MyTask(Task): def run(self, **kwargs): CreateContainer().run() StartContainer().run() WaitOnContainer().run() ``` (note this is pseudocode!) – chriswhite Jun 23 '20 at 16:45
  • @chriswhite I tried, but I always to some error like "no flow context found". Not sure if tasks can really be merged like that. – theDmi Jun 24 '20 at 14:21
  • 1
    With prefect 2.2.0 this is no longer true. Can you provide this example but for prefect v2 @Josh – lamhoangtung Aug 29 '22 at 05:58