dask.distributed不使用集群

Question

我无法使用分布式集群处理这个块。

import pandas as pd
from dask import dataframe as dd 
import dask

df = pd.DataFrame({'reid_encod': [[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10]]})
dask_df = dd.from_pandas(df, npartitions=3)
save_val = []
def add(dask_df):
    for _, outer_row in dask_df.iterrows():
        for _, inner_row in dask_df.iterrows():
            for base_encod in outer_row['reid_encod']:
               for compare_encod in inner_row['reid_encod']:
                   val = base_encod + compare_encod
                   save_val.append(val)
    return save_val

from dask.distributed import Client

client = Client(...)
dask_compute = dask.delayed(add)(dask_df)
dask_compute.compute()

另外，我也很少有疑问。

1. dask.delayed是否使用可用的集群来进行计算。
2. 我能不能用延迟的方法来模仿这个熊猫DF的循环迭代，并使用集群中的多台计算机来进行计算。
3. dask.distributed在熊猫数据方面有工作吗？
4. 我们可以在dask.delayed中使用dask.distributed吗？
5. 如果上面的编程方法是错误的，您能指导我是为上面的场景选择延迟还是dask。

Answer 1

为了记录在案，我要回答一些问题，尽管我想指出我先前关于这个问题的一般性观点。

dask.delayed是否使用可用的集群来进行计算。

如果您已经为分布式集群创建了一个客户端，dask将使用它进行计算，除非您另有指定。

我能不能用延迟的方法来模仿这个熊猫DF的循环迭代，并使用集群中的多台计算机来进行计算。

是的，如果你愿意的话，你可以在一般情况下使用延迟与熊猫的数据并行。但是，您的dataframe只有一行，因此在本例中并不明显--它取决于您真正想要实现的目标。

dask.distributed在熊猫数据方面有工作吗？

是的，您可以使用分布式处理python所能做的任何事情，因为它只是执行代码的python进程。它是否给你带来了你想要的表演是另一个问题。

我们可以在dask.delayed中使用dask.distributed吗？

是的，分布式可以执行dask通常能够执行的任何事情，包括延迟的函数/对象。

如果上面的编程方法是错误的，您能指导我是为上面的场景选择延迟还是dask。

这不是一件容易的事情，我不清楚这是否是一个数据操作。它看起来更像是一个数组--但是，我再次注意到，您的函数实际上根本没有返回任何有用的东西。

在本教程中：把熊猫的数据传递给延迟；与dataframe API相同。

Answer 2

代码的主要问题是在本节中勾画出最佳实践:不要将Dask集合传递给延迟函数。这意味着，您应该使用delayed API或dataframe API。虽然您可以转换数据格式<->延迟，但不建议像这样简单地传递。

此外，

• 您的dataframe中只有一行，所以您只能得到一个分区，而没有任何并行性。你只能让事情慢下来。
• 这似乎是一个所有的操作(N^2)，所以如果您有许多行(Dask的正常情况)，那么无论您使用了多少核，它大概都要花费非常长的时间。
• 在熊猫行中传递列表不是一个好主意，也许你想使用数组？
• 这个函数没有返回任何有用的东西，所以它根本不清楚你想要实现什么。在MVCE的描述下，您将看到对“预期结果”和“哪里出错”的引用。为了得到更多的帮助，请更加精确。