为什么我的Dask期货卡在“待定"状态而从未完成? [英] Why do my Dask Futures get stuck in 'pending' and never finish?

问题描述

我有一些长时间运行的代码(大约需要5-10分钟的处理时间)，我想以Dask Future 的方式运行.我可以将其作为一个函数运行，这是一系列几个离散的步骤:

I have some long-running code (~5-10 minute processing) that I'm trying to run as a Dask Future. It's a series of several discrete steps that I can either run as one function:

result : Future = client.submit(my_function, arg1, arg2)

或者我可以分为几个中间步骤:

Or I can split up into intermediate steps:

# compose the result from the same intermediate results but with Futures
intermediate1 = client.submit(my_function1, arg1)
intermediate2 = client.submit(my_function2, arg1, arg2)
intermediate3 = client.submit(my_function3, intermediate2, arg1)
result = client.submit(my_function4, intermediate3)

如果我在本地运行此命令(例如， result = my_function(arg1，arg2))，则此操作完成.如果我将其提交给Dask，则可以按预期立即将我的 Future 退回给我，但该工作永远无法完成.此外，如果我将 result.key 用作跟踪作业状态的方法，则以后将Future重构为 result = Future(key)，它是始终处于待处理状态.

If I run this locally (eg, result = my_function(arg1, arg2)), it completes. If I submit it to Dask, I immediately get my Future back - as expected - but the job never completes. Further, if I grab the result.key as a way to track the status of the job, later reconstructing the future as result = Future(key), it always has a state of pending.

我想首先按原样运行它，这样我就可以将我的处理工作转移给我的Dask工作者，而不是将处理请求的API转移给我，然后我希望能够开始跨节点拆分工作，所以我可以提高性能.但是，为什么我的工作只是蒸蒸日上?查看我的Dask Scheduler Web界面，似乎没有显示作业.但是我知道Dask正在工作，因为我可以从Jupyter笔记本中向其中提交代码.

I want to first get it running as-is so that I can have my processing offloaded to my Dask workers instead of an API that's handling the requests, and then I want to be able to start splitting up work across nodes so I can improve the performance. But why are my jobs just evaporating? Looking at my Dask scheduler web interface, it doesn't appear the jobs are even showing up. But I know Dask is working because I can submit code to it from my Jupyter notebook.

我正在从Flask服务器中调用 client.submit ，然后返回密钥，以便以后使用.大概是:

I'm calling client.submit from a Flask server, and I'm returning the key so it can be used later. Roughly:

@app.route('/submit')
def submit():
    # ...
    future = client.submit(my_function, arg1, arg2)
    return jsonify({"key": future.key})

@app.route('/status/<key>')
def status(key):
    future = Future(key)
    return jsonify({"status": future.status})

当我的应用程序部署到Kubernetes时，我的/submit 路由会获得Future键，但是我的Dask状态页面没有显示任何处理任务.如果我在本地运行Flask，则确实会看到一个任务显示，并且在预期的延迟之后会显示我的作业的输出 .但是，当我用从/submit 返回的Future键命中自己的/status/< key> 路径时，它始终显示状态为待处理.

When my application is deployed to Kubernetes, my /submit route gets a Future key back, but my Dask status page doesn't show any processing task. If I run Flask locally, I do see a task show up, and the output of my job does show up after an expected delay; however, when I hit my own /status/<key> path with the Future key returned from /submit, it always shows the state is pending.

推荐答案

如果所有指向任务的期货都消失了，那么Dask可以随意忘记该任务.这使Dask可以清理工作，而不是让所有中间结果永远存在.

If all futures pointing to a task disappear then Dask feels free to forget about that task. This allows Dask to clean up work, rather than have all intermediate results stay around forever.

如果要保留参考，则需要保留期货.这告诉Dask您仍然在意结果.您可以在烧瓶应用中通过创建字典在本地进行此操作.

If you want to hold on to references then you'll need to hold on to futures. This tells Dask that you still care about the result. You can do this locally in your flask app by creating a dictionary.

futures = {}

@app.route('/submit')
def submit():
    # ...
    future = client.submit(my_function, arg1, arg2)
    futures[future.key] = future
    return jsonify({"key": future.key})

@app.route('/status/<key>')
def status(key):
    future = futures[key]
    return jsonify({"status": future.status})

但是您还需要考虑何时可以清理并释放这些期货.通过这种方法，您将慢慢填满您的记忆.

But you'll also want to think about when you can clean up and release those futures. With this approach you will slowly fill up your memory.

这篇关于为什么我的Dask期货卡在“待定"状态而从未完成?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！