python - 将分类列转换为附加列

Question

我有一个以前从 avro 文件加载的以下数据框形式的大型数据集

<表类="s-表"><头>时间戳id类别值<正文>2021-01-01 00:00:00+00:00一个dg2021-01-01 00:10:00+00:00一个dh2021-01-01 00:10:00+00:00一个eh2021-01-01 00:00:00+00:00beh

我想旋转 category列(包含大约 50 个不同类别的顺序)并沿着 timestamp 进行重复数据删除。和 id列所以结果看起来像这样

<表类="s-表"><头>id时间戳de<正文>一个2021-01-01 00:00:00+00:00g南一个2021-01-01 00:10:00+00:00hhb2021-01-01 00:00:00+00:00南h

我知道如何在 pandas 中实现这一目标将多指数与 stack 一起使用/unstack操作，但是我的数据集太大而无法使用 pandas无需手动批处理和dask不支持多指数。有什么方法可以用 dask 有效地完成吗？ ？

编辑:

正如@Dahn 所指出的，我用 pandas 创建了一个最小的合成示例:

import pandas as pd

records = [
    {'idx': 0, 'id': 'a', 'category': 'd', 'value': 1},
    {'idx': 1, 'id': 'a', 'category': 'e', 'value': 2},
    {'idx': 2, 'id': 'a', 'category': 'f', 'value': 3},
    {'idx': 0, 'id': 'b', 'category': 'd', 'value': 4},
    {'idx': 1, 'id': 'c', 'category': 'e', 'value': 5},
    {'idx': 2, 'id': 'c', 'category': 'f', 'value': 6}
]

frame = pd.DataFrame(records)

   idx id category  value
0    0  a        d      1
1    1  a        e      2
2    2  a        f      3
3    0  b        d      4
4    1  c        e      5
5    2  c        f      6

frame = frame.set_index(['id', 'idx', 'category'], drop=True).unstack().droplevel(0, axis=1).reset_index()
frame.columns.name = ''

  id  idx    d    e    f
0  a    0  1.0  NaN  NaN
1  a    1  NaN  2.0  NaN
2  a    2  NaN  NaN  3.0
3  b    0  4.0  NaN  NaN
4  c    1  NaN  5.0  NaN
5  c    2  NaN  NaN  6.0

Answer 1

我不认为 Dask 会在 2021 年 10 月实现这一点。这可能是因为不支持 unstack 需要的多索引。已经有some work on this不过最近。

但是，我认为这仍然可以使用 apply-concat-apply paradigm 实现(和 apply_concat_apply 函数)。

下面的解决方案适用于您提供的示例，原则上，我认为它应该普遍适用，但我不确定。请谨慎行事，如果可能，请检查结果是否与 Pandas 给您的结果一致。我也将其发布为 feature request在 Dask 的 github 上。

import dask.dataframe as dd

# Create Dask DataFrame out of your `frame`
# npartitions is more than 1 to demonstrate this works on a partitioned datataset
df = dd.from_pandas(frame, npartitions=3)

# Dask needs to know work out what the categories are
# Alternatively you can use df.categorize
# See https://docs.dask.org/en/latest/dataframe-categoricals.html
category = 'category'
df[category] = df[category].astype(category).cat.as_known()

# Dask needs to know what the resulting DataFrame looks like
new_columns = pd.CategoricalIndex(df[category].cat.categories, name=category)
meta = pd.DataFrame(columns=new_columns, 
                    index=df._meta.set_index(['idx', 'id']).index)

# Implement using apply_concat_apply ("aca")
# More details: https://blog.dask.org/2019/10/08/df-groupby
def identity(x): return x

def my_unstack(x):
    return x.set_index(['id', 'idx', 'category'], drop=True).unstack()
    
def combine(x):
    return x.groupby(level=[0, 1]).sum()

result = dd.core.apply_concat_apply([df], 
                   chunk=identity, 
                   aggregate=my_unstack, 
                   combine=combine,
                   meta=meta)

result.compute()

选项 B:map_partitions

如果您已经能够至少根据 idx 或 id 之一对数据进行排序，那么您也可以简单地使用 map_partitions并将每个分区视为一个 Pandas 数据框。

这应该会显着改善内存使用和整体性能。

# df has sorted index `idx` in this scenario

category = 'category'
existing_categories = df[category].astype(category).cat.as_known().cat.categories
categories = [('value', cat) for cat in existing_categories]

new_columns = pd.MultiIndex.from_tuples(categories, names=(None, category))

meta = pd.DataFrame(columns=new_columns, 
                    index=df._meta.set_index(['idx', 'id']).index)

def unstack_add_columns(x):
    x = x.set_index(['id', 'category'], append=True, drop=True).unstack()
    # make sure that result contains all necessary columns
    return x.reindex(columns=new_columns) 

df.map_partitions(unstack_add_columns, meta=meta)

如果你不能保证 idx 会被排序，你可以尝试类似的东西

df_sorted = df.set_index('idx')
# I recommend saving to disk in between set_index and the rest
df_sorted.to_parquet('data-sorted.parq')

但这本身可能会带来内存问题。