如何在 Dask 中进行行处理和项目分配

Question

类似的未回答问题：Row by row processing of a Dask DataFrame

我正在处理数百万行长的数据帧，所以现在我正在尝试并行执行所有数据帧操作。我需要转换为 Dask 的这样一个操作是：

 for row in df.itertuples():                                                                                                                                                                                                         
     ratio = row.ratio                                                                                                                                                                                                                     
     tmpratio = row.tmpratio                                                                                                                                                                                                                                                                                                                                                                                                 
     tmplabel = row.tmplabel                                                                                                                                                                                                               
     if tmpratio > ratio:                                                                                                                                                                                                                  
         df.loc[row.Index,'ratio'] = tmpratio                                                                                                                                                                                        
         df.loc[row.Index,'label'] = tmplabel

在 Dask 中按索引设置值或有条件地在行中设置值的合适方法是什么？鉴于 .loc 在 Dask 中不支持项目分配，因此在 Dask 中似乎也没有 set_value、at[] 或 iat[]。

我尝试使用 map_partitions with assign，但我没有看到任何在行级执行条件赋值的能力。

Answer 1

Dask 数据框不支持有效的迭代或行分配。一般来说，这些工作流程很少能很好地扩展。它们在 Pandas 本身也很慢。

相反，您可以考虑使用 Series.where 方法。这是一个最小的例子：

In [1]: import pandas as pd

In [2]: df = pd.DataFrame({'x': [1, 2, 3], 'y': [3, 2, 1]})

In [3]: import dask.dataframe as dd

In [4]: ddf = dd.from_pandas(df, npartitions=2)

In [5]: ddf['z'] = ddf.x.where(ddf.x > ddf.y, ddf.y)

In [6]: ddf.compute()
Out[6]:
   x  y  z
0  1  3  3
1  2  2  2
2  3  1  3