prefect-saturn

prefect-saturn 是一个Python包，它使得在 Prefect Cloud 上使用 Saturn Cloud 运行Dask集群中的流程变得容易。有关详细教程，请参阅 "使用Prefect Cloud实现容错数据管道 "。

安装

prefect-saturn 可在PyPi上找到。

pip install prefect-saturn

prefect-saturn 可以直接从GitHub安装

pip install git+https://github.com/saturncloud/prefect-saturn.git@main

入门

prefect-saturn 旨在在 Saturn Cloud 环境中使用，例如Jupyter笔记本。

import prefect
from prefect import Flow, task
from prefect_saturn import PrefectCloudIntegration


@task
def hello_task():
    logger = prefect.context.get("logger")
    logger.info("hello prefect-saturn")


flow = Flow("sample-flow", tasks=[hello_task])

project_name = "sample-project"
integration = PrefectCloudIntegration(
    prefect_cloud_project_name=project_name
)
flow = integration.register_flow_with_saturn(flow)

flow.register(
    project_name=project_name,
    labels=["saturn-cloud"]
)

自定义Dask

您可以根据需要自定义运行prefect流程的Dask集群的大小和行为。prefect_saturn.PrefectCloudIntegration.register_flow_with_saturn() 接受两个参数来完成此操作

• dask_cluster_kwargs：传递给构造函数 dask_saturn.SaturnCluster 的关键字参数。
• dask_adapt_kwargs：用于配置 "自适应扩展" 的关键字参数

例如，下面的代码告诉土星（Saturn）这个流程应该在具有3个xlarge工作节点的Dask集群上运行，并且一旦流程运行完成，prefect应该关闭集群。

flow = integration.register_flow_with_saturn(
    flow=flow,
    dask_cluster_kwargs={
        "n_workers": 3,
        "worker_size": "xlarge",
        "autoclose": True
    }
)

flow.register(
    project_name=project_name,
    labels=["saturn-cloud"]
)

贡献

请参阅CONTRIBUTING.md，了解如何测试和对prefect-saturn进行贡献。