dask DataFrame.assign炸毁了dask图

Question

所以我有一个dask DataFrame.append的问题.我从主数据中生成了许多衍生特性,并将它们附加到主数据框中.之后,任何一组列的dask图表都会被炸毁.这是一个小例子:

%pylab inline
import numpy as np
import pandas as pd
import dask.dataframe as dd
from dask.dot import dot_graph

df=pd.DataFrame({'x%s'%i:np.random.rand(20) for i in range(5)})
ddf = dd.from_pandas(df, npartitions=2)

dot_graph(ddf['x0'].dask)

这是预期的dask图

g=ddf.assign(y=ddf['x0']+ddf['x1'])
dot_graph(g['x0'].dask)

这里同一列的图表以不相关的计算方式展开

想象一下,我有很多产生的列.因此,任何特定列的计算图包括所有其他列的无关计算.即在我的情况下,我有len(ddf ['someColumn'].dask)> 100000.因此很快就会无法使用.

所以我的问题是这个问题可以解决吗？有没有现成的方法呢？如果不是 - 我应该朝哪个方向实施这个目标？

谢谢!

Answer 1

您可能希望构建多个dask系列,然后在最后将它们连接在一起,而不是将新列连续分配给dask数据帧

所以不要这样做:

df['x'] = df.w + 1
df['y'] = df.x * 10
df['z'] = df.y ** 2

做这个

x = df.w + 1
y = x + 10
z = y * 2
df = df.assign(x=x, y=y, z=z)

或这个:

dd.concat([df, x, y, z], axis=1)

但是,这仍然可能导致图表中的任务数量相同,但可能会导致更少的内存副本.

或者,如果所有转换都是逐行的,那么您可以构建一个pandas函数并将其映射到所有分区

def f(part):
    part = part.copy()
    part['x'] = part.w + 1
    part['y'] = part.x * 10
    part['z'] = part.y ** 2
    return part

df = df.map_partitions(f)

此外,虽然百万节点的任务图不太理想,但也应该没问题.我看到更大的图表运行得舒服.