我已经在GCP上部署了一个Kubernetes集群，结合了prefect和dask。作业在正常情况下运行良好，但无法缩放 2 倍的数据。到目前为止，我已经将其缩小到由于内存使用率高而关闭的调度程序。 达斯克调度程序内存 一旦内存使用量达到 2GB，作业就会失败并显示"未检测到检测信号"错误。
有一个单独的构建 python 文件可用，我们可以在其中设置工作线程内存和 CPU。有一个 dask 网关包，我们可以在其中获取网关选项并设置工作线程内存。
options.worker_memory = 32
options.worker_cores = 10
cluster = gateway.new_cluster(options)
cluster.adapt(minimum=4, maximum=20)
我无法弄清楚在哪里以及如何增加 dask 调度程序的内存分配。
Specs:
Cluster Version: 1.19.14-gke.1900
Machine type - n1-highmem-64
Autoscaling set to 6 - 1000 nodes per zone
all nodes are allocated 63.77 CPU and 423.26 GB
首先要解释为什么会出现 Flow 心跳： 省长使用检测信号来检查您的流是否仍在运行。如果 Prefect 没有检测信号，则在远程执行环境(如 Kubernetes 作业)上失去通信并死亡的流将在 UI 中永久显示为"正在运行"。通常，"未检测到检测信号"是由于内存不足或流执行长时间运行的作业而发生的。
您可以尝试的一种解决方案是在运行配置上设置以下环境变量 - 这将更改从进程到线程的检测信号行为，并可以帮助解决问题：
from prefect.run_configs import UniversalRun
flow.run_config = UniversalRun(env={"PREFECT__CLOUD__HEARTBEAT_MODE": "thread"})
正如你提到的，最好的解决方案是增加Dask工作人员的内存。如果使用长时间运行的群集，则可以按以下方式进行设置：
dask-worker tcp://scheduler:port --memory-limit="4 GiB"
如果您将集群类传递给您的 Dask 执行器，例如coiled.Cluster，您可以同时设置：
- scheduler_memory - 默认为 4 GiB
- worker_memory - 默认为 8 GiB
以下是在流中设置它的方法：
import coiled
from prefect import Flow
from prefect.executors import DaskExecutor
flow = Flow("test-flow")
executor = DaskExecutor(
cluster_class=coiled.Cluster,
cluster_kwargs={
"software": "user/software_env_name",
"shutdown_on_close": True,
"name": "prefect-cluster",
"scheduler_memory": "4 GiB",
"worker_memory": "8 GiB",
},
)
flow.executor = executor