dask。分布式未利用群集

Question

bounty到期在3天内。该问题的答案有资格获得+ 100声望赏金。Naren想要引起更多关注来回答这个问题。

我无法使用分布式集群来处理此块。

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的for循环迭代，并使用群集中存在的多台计算机进行计算。

3. 在熊猫数据框上进行分布式操作。

4. 我们可以使用dask.delayed在dask.distributed中。

5. 如果上述编程方法不正确，您可以指导我为上述情况选择延迟还是慢速DF。

Answer 1

您的代码的主要问题已在最佳实践的this section中进行了概述：不要将Dask集合传递给延迟的函数。这意味着，您应该使用delayed API 或dataframeAPI。虽然您可以转换延迟的数据帧，但不建议像这样简单地传递。此外，

您在数据帧中只有一行，因此您只会得到一个分区，并且没有并行性。您只能像这样放慢速度。
• 这似乎是一个万能的操作（N ^ 2），因此，如果您有很多行（Dask的正常情况），那么不管您使用了多少个内核，都将花费极长的时间]
• 在熊猫行中传递列表不是一个好主意，也许您想使用数组？
• 该函数不会返回任何有用的信息，因此根本不清楚您要实现的目标。在MVCE的描述下，您将看到对“预期结果”和“出了什么问题”的引用。要获得更多帮助，请更精确。

Answer 2

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