在 dask DF 中跨多个列展开类似列表的列答案

【问题标题】：Expand a list-like column in dask DF across several columns在 dask DF 中跨多个列展开类似列表的列
【发布时间】：2021-11-20 23:08:30
【问题描述】：

这与之前关于如何将基于列表的列扩展到多个列的问题类似，但我看到的解决方案似乎不适用于 Dask。请注意，我正在使用的真正 DF 太大而无法保存在内存中，因此不能先转换为 pandas。

我有一个包含列表的列：

df = pd.DataFrame({'a': [np.random.randint(100, size=4) for _ in range(20)]})
dask_df = dd.from_pandas(df, chunksize=10)
dask_df['a'].compute()

0     [52, 38, 59, 78]
1     [79, 71, 13, 63]
2     [15, 81, 79, 76]
3      [53, 4, 94, 62]
4     [91, 34, 26, 92]
5      [96, 1, 69, 27]
6     [84, 91, 96, 68]
7     [93, 56, 45, 40]
8      [54, 1, 96, 76]
9      [27, 11, 79, 7]
10    [27, 60, 78, 23]
11    [56, 61, 88, 68]
12    [81, 10, 79, 65]
13     [34, 49, 30, 3]
14    [32, 46, 53, 62]
15    [20, 46, 87, 31]
16      [89, 9, 11, 4]
17    [26, 46, 19, 27]
18    [79, 44, 45, 56]
19    [22, 18, 31, 90]
Name: a, dtype: object

根据这个solution，如果这是一个 pd.DataFrame 我可以这样做：

new_dask_df = dask_df['a'].apply(pd.Series)

ValueError: The columns in the computed data do not match the columns in the provided metadata
Extra:   [1, 2, 3]
Missing: []

这里列出了另一种解决方案：

import dask.array as da
import dask.dataframe as dd
x = da.ones((4, 2), chunks=(2, 2))
df = dd.io.from_dask_array(x, columns=['a', 'b'])
df.compute()

所以对于 dask 我尝试了：

df = dd.io.from_dask_array(dask_df.values)

但这只是吐出与我之前相同的 DF： [1]：https://i.stack.imgur.com/T099A.png

不太清楚为什么示例 'x' 和我的 df 中的值之间的类型是相同的：

print(type(dask_df.values), type(x))
<class 'dask.array.core.Array'> <class 'dask.array.core.Array'>
print(type(dask_df.values.compute()[0]), type(x.compute()[0]))
<class 'numpy.ndarray'> <class 'numpy.ndarray'>

编辑：我有一个可行的解决方案，但它涉及遍历每个 groupby 对象。感觉应该有更好的办法：

dask_groups = dask_df.explode('a').reset_index().groupby('index')
final_df = []
for idx in dask_df.index.values.compute():
    group = dask_groups.get_group(idx).drop(columns='index').compute()
    group_size = list(range(len(group)))
    row = group.transpose()
    row.columns = group_size
    row['index'] = idx
    final_df.append(dd.from_pandas(row, chunksize=10))
final_df = dd.concat(final_df).set_index('index')

【问题讨论】：

标签： python pandas dask

【解决方案1】：

在这种情况下，dask 不知道结果是什么，所以最好明确指定meta：


# this is a short-cut to use the existing pandas df
# in actual code it is sufficient to provide an
# empty series with the expected dtype
meta = df['a'].apply(pd.Series)

new_dask_df = dask_df['a'].apply(pd.Series, meta=meta)
new_dask_df.compute()

【讨论】：

这在技术上有效，但它假设我可以访问 df.在我的真实程序中，df 不存在，一切都作为 dask df 存在，所以为了创建元对象，我必须从计算 dask_df 开始，这违背了目的。
不完全是，meta 不需要包含数据，只需要包含列名和数据类型。
谢谢，这行得通：columns = list(range(4)) meta = pd.DataFrame(columns=columns) for x in columns: meta[x] = meta[x].astype(int ) new_dask_df = dask_df['a'].apply(pd.Series, meta=meta)

【解决方案2】：

我有一个可行的解决方案。我的原始函数创建了一个列表，该列表导致列表列，如上所述。更改应用函数以返回一个 dask bag 似乎可以解决问题：

def create_df_row(x):
    vals = np.random.randint(2, size=4)
    return db.from_sequence([vals], partition_size=2).to_dataframe()

test_df = dd.from_pandas(pd.DataFrame({'a':[random.choice(['a', 'b', 'c']) for _ in range(20)]}), chunksize=10)
test_df.head()

mini_dfs = [*test_df.groupby('a')['a'].apply(lambda x: create_df_row(x))]
result = dd.concat(mini_dfs)
result.compute().head()

但不确定这是否解决了内存中的问题，因为现在我持有一个 groupby 结果列表。

【讨论】：

【解决方案3】：

以下是手动跨多个列展开类似列表的列的方法：

dask_df["a0"] = dask_df["a"].str[0]
dask_df["a1"] = dask_df["a"].str[1]
dask_df["a2"] = dask_df["a"].str[2]
dask_df["a3"] = dask_df["a"].str[3]

print(dask_df.head())

                  a  a0  a1  a2  a3
0   [71, 16, 0, 10]  71  16   0  10
1  [59, 65, 99, 74]  59  65  99  74
2  [83, 26, 33, 38]  83  26  33  38
3   [70, 5, 19, 37]  70   5  19  37
4    [0, 59, 4, 80]   0  59   4  80

SultanOrazbayev 的回答似乎更优雅。

【讨论】：