将新列附加到 dask 数据框答案

【问题标题】：Appending new column to dask dataframe将新列附加到 dask 数据框
【发布时间】：2018-04-06 00:05:05
【问题描述】：

我有一个现有的 dask 数据框 df 我希望在其中执行以下操作：

df['rand_index'] = np.random.permutation(len(df))

但是，这会产生错误，Column assignment doesn't support type ndarray。我尝试使用df.assign(rand_index = np.random.permutation(len(df))，它给出了同样的错误。

这是一个最小（非）工作示例：

import pandas as pd
import dask.dataframe as dd
import numpy as np

df = dd.from_pandas(pd.DataFrame({'A':[1,2,3]*10, 'B':[3,2,1]*10}), npartitions=10)
df['rand_index'] = np.random.permutation(len(df))

注意：

使用df = df.map_partitions(add_random_column_to_pandas_dataframe, ...) 提到的上一个问题，但我不确定这是否与这个特殊情况有关。

编辑 1

我尝试过 df['rand_index'] = dd.from_array(np.random.permutation(len_df)) 其中，执行没有问题。当我检查df.head() 时，似乎新列创建得很好。但是，当我查看df.tail() 时，rand_index 是一堆NaNs。

实际上只是为了确认我检查了df.rand_index.max().compute()，结果证明它小于len(df)-1。所以这可能是df.map_partitions 发挥作用的地方，因为我怀疑这是 dask 被分区的问题。在我的特殊情况下，我有 80 个分区（不是指示例案例）。

【问题讨论】：

标签： python dask

【解决方案1】：

您需要将 np.random.permutation(len(df)) 转换为 dask 可以理解的类型：

permutations = dd.from_array(np.random.permutation(len(df)))
df['rand_index'] = permutations
df

这将产生：

Dask DataFrame Structure:
                    A      B rand_index
npartitions=10                         
0               int64  int64      int32
3                 ...    ...        ...
...               ...    ...        ...
27                ...    ...        ...
29                ...    ...        ...
Dask Name: assign, 61 tasks

所以现在你要不要.compute()来计算实际结果。

【讨论】：

如果索引不相同（例如，原始 df 具有日期时间索引而新的 Series 具有 int 索引），则会出现问题

【解决方案2】：

要分配列，您应该使用df.assign

【讨论】：

恐怕这行不通。做df = df.assign(rand_index2=dd.from_array(np.random.permutation(len_df))) 仍然给了我NaNs 在df.tail()
您使用的是哪个 dask 版本？ df = df.assign(rand_index=dd.from_array(np.random.permutation(len(df)))) 为我工作。
版本0.15.3

【解决方案3】：

遇到了与编辑 1 中相同的问题。

我的解决方法是从现有数据框中获取一个唯一列，并将其输入到要附加的数据框中。

import dask.dataframe as dd
import dask.array as da
import numpy as np
import panda as pd

df = dd.from_pandas(pd.DataFrame({'A':[1,2,3]*2, 'B':[3,2,1]*2, 'idx':[0,1,2,3,4,5]}), npartitions=10)
chunks = tuple(df.map_partitions(len).compute())
size = sum(chunks)
permutations = da.from_array(np.random.permutation(len(df)), chunks=chunks)
idx = da.from_array(df['idx'].compute(), chunks=chunks)
ddf = dd.concat([dd.from_dask_array(c) for c in [idx,permutations]], axis = 1)
ddf.columns = ['idx','rand_idx']
df = df.merge(ddf, on='idx')
df = df.set_index('rand_idx')
df.compute().head()

【讨论】：