将 Dask Dataframe 中的列拆分为 n 列

Question

在 Dask 数据框中的一列中，我有这样的字符串：

column_name_1	column_name_2
a^b^c	j
e^f^g	k^l
h^i	m

我需要将这些字符串拆分为以下列： 数据框，就像

column_name_1column_name_2column_name_1_1column_name_1_2column_name_1_3column_name_2_1column_name_2_2a	b	^	c	的	相同	这样
^	j	a	b	c	j
e^f^g	k^l	e	f	g	k	l
h^i	m	h	i		m

如果事先不知道数据中出现了多少次分隔符，我无法弄清楚如何执行此操作。此外，数据框中有数十列需要单独保留，因此我需要能够像这样指定要拆分的列。

我最大的努力要么包括类似的东西

df[["column_name_1_1","column_name_1_2 ","column_name_1_3"]] = df["column_name_1"].str.split('^',n=2, expand=True)

但它失败了

ValueError：计算数据中的列与提供的元数据中的列不匹配

Answer 1

这里有 2 个解决方案，无需 stack，但对选定的列名称使用循环：

cols = ['column_name_1','column_name_2']
for c in cols:
    df = df.join(df[c].str.split('^',n=2, expand=True).add_prefix(f'{c}_').fillna(''))

print (df)
  column_name_1 column_name_2 column_name_1_0 column_name_1_1 column_name_1_2  \
0         a^b^c             j               a               b               c   
1         e^f^g           k^l               e               f               g   
2           h^i             m               h               i                   

  column_name_2_0 column_name_2_1  
0               j                  
1               k               l  
2               m                  

或者修改另一个解决方案：

cols = ['column_name_1','column_name_2']
dfs = [df[c].str.split('^',n=2, expand=True).add_prefix(f'{c}_').fillna('') for c in cols]
df = pd.concat([df] + dfs, axis=1)
print (df)
  column_name_1 column_name_2 column_name_1_0 column_name_1_1 column_name_1_2  \
0         a^b^c             j               a               b               c   
1         e^f^g           k^l               e               f               g   
2           h^i             m               h               i                   

  column_name_2_0 column_name_2_1  
0               j                  
1               k               l  
2               m                  

Answer 2

不幸的是使用 dask.dataframe.Series Dask 尚不支持带有 expand=True 的 .str.split 和未知数量的分割，以下返回 NotImplementedError：

import dask.dataframe as dd
import pandas as pd

ddf = dd.from_pandas(
    pd.DataFrame({
        'column_name_1': ['a^b^c', 'e^f^g', 'h^i'], 'column_name_2': ['j', 'k^l', 'm']
    }), npartitions=2
)

# returns NotImplementedError
ddf['column_name_1'].str.split('^', expand=True).compute()

通常当一个pandas 等效项尚未在 Dask 中实现，map_partitions 可用于在每个 DataFrame 分区上应用 Python 函数。然而，在这种情况下，Dask 仍然需要知道需要多少列才能延迟生成 Dask DataFrame，如 meta 参数。这使得使用 Dask 来完成这项任务具有挑战性。相关地，发生 ValueError 是因为 column_name_2 仅需要 1 次分割，并返回具有 2 列的 Dask DataFrame，但 Dask 期望具有 3 列的 DataFrame。

如果您提前知道分割数，这里有一个解决方案（根据 @Fontanka16 的答案构建）：

import dask.dataframe as dd
import pandas as pd

ddf = dd.from_pandas(
    pd.DataFrame({
        'column_name_1': ['a^b^c', 'e^f^g', 'h^i'], 'column_name_2': ['j', 'k^l', 'm']
    }), npartitions=2
)

ddf_list = []
num_split_dict = {'column_name_1': 2, 'column_name_2': 1}
for col, num_splits in num_split_dict.items():
    split_df = ddf[col].str.split('^', n=num_splits, expand=True).add_prefix(f'{col}_')
    ddf_list.append(split_df)
new_ddf = dd.concat([ddf] + ddf_list, axis=1)
new_ddf.compute()