docker - 如何在 Prefect 流程中使用自定义 Docker 存储？

Question

我已经设置了一个 Dask 集群，并且我很高兴向它发送基本的 Prefect 流。 现在我想做一些更有趣的事情，获取一个带有我的 python 库的自定义 docker 镜像，并在 dask 集群上执行流程/任务。

我的假设是我可以将 dask 集群(调度程序和工作人员)保留在自己的 python 环境中(在检查了所有各种消息传递库到处都有匹配的版本之后)。也就是说，如果流程在我的自定义存储中执行，我预计不需要将我的库添加到这些机器。 但是，要么我没有正确设置存储，要么假设上述情况不安全。换句话说，也许在腌制我的自定义库中的对象时，Dask 集群确实需要了解我的 python 库。假设我有一些通用的 python 库，名为 data...

import prefect    
from prefect.engine.executors import DaskExecutor
#see https://docs.prefect.io/api/latest/environments/storage.html#docker
from prefect.environments.storage import Docker

#option 1
storage = Docker(registry_url="gcr.io/my-project/",
                 python_dependencies=["some-extra-public-package"],
                 dockerfile="/path/to/Dockerfile")
#this is the docker build and register workflow!
#storage.build()

#or option 2, specify image directly
storage = Docker(
        registry_url="gcr.io/my-project/", image_name="my-image", image_tag="latest"
    )

#storage.build()

def get_tasks():
    return [
        "gs://path/to/task.yaml"
           ]

@prefect.task
def run_task(uri):
    #fails because this data needs to be pickled ??
    from data.tasks import TaskBase
    task =  TaskBase.from_task_uri(uri)
    #task.run()
    return "done"

with prefect.Flow("dask-example",
                 storage = storage) as flow:
    #chain stuff...
    result =  run_task.map(uri=get_tasks())

executor = DaskExecutor(address="tcp://127.0.01:8080")
flow.run(executor=executor)

谁能解释一下这种基于 docker 的工作流程应该如何/是否有效？

Answer 1

您的 dask 工作人员将需要访问您的任务运行所依赖的相同 python 库。实现此目的的最简单方法是使用与 Flow 相同的镜像来运行 dask 工作程序。您可以手动执行此操作，或使用类似 DaskCloudProviderEnvironment 的内容这将自动使用相同的镜像创建每个流程运行的短期 Dask 集群。