dask groupby 应用然后合并回数据框答案

【问题标题】：dask groupby apply then merge back to dataframedask groupby 应用然后合并回数据框
【发布时间】：2017-12-06 20:10:31
【问题描述】：

我会创建一个新列，它是 groupby 的结果并应用另一列，同时保持数据框的顺序（或至少能够将其排序回来）。

示例：我想按组规范化信号列

import dask
import numpy as np
import pandas as pd
from dask import dataframe

def normalize(x):
    return ((x - x.mean())/x.std())


data = np.vstack([np.arange(2000), np.random.random(2000), np.round(np.linspace(0, 10, 2000))]).T
df = dataframe.from_array(data, columns=['index', 'signal', 'id_group'], chunksize=100)
df = df.set_index('index')

normalized_signal =  df.groupby('id_group').signal.apply(normalize, meta=pd.Series(name='normalized_signal_by_group'))
normalized_signal.compute()

我确实得到了正确的系列，但索引被打乱了。我要在数据框中恢复这个系列吗？

我试过了

df['normalized_signal'] = normalized_signal
df.compute()

但我明白了

ValueError：并非所有分区都是已知的，无法对齐分区。请使用set_index设置索引。

我也尝试了合并，但我的最终数据帧最终被打乱了，没有简单的方法可以沿着索引求助

df2 = df.merge(normalized_signal.to_frame(), left_index=True, right_index=True, how='left')
df2.compute()

当我在 pandas 中计算序列而不是 sort_index() 时它可以工作，但这似乎效率不高。

df3 = df.merge(normalized_signal.to_frame().compute().sort_index(), left_index=True, right_index=True, how='left')
df3.compute()

等效的熊猫方式是：

df4 = df.compute()
df4['normalized_signal_by_group'] = df4.groupby('id_group').signal.transform(normalize)
df4

【问题讨论】：

是否会使用“延迟”来查询数据库的分组块，做数学计算，而不是使用“延迟”是一个有效的解决方案？

标签： python dask

【解决方案1】：

不幸的是，transform 尚未在 dask 中实现。我的（丑陋的）解决方法是：

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

pd.options.mode.chained_assignment = None

def normalize(x):
    return ((x - x.mean())/x.std())

def dask_norm(gp):
    gp["norm_signal"] = normalize(gp["signal"].values)
    return(gp.as_matrix())

data = np.vstack([np.arange(2000), np.random.random(2000), np.round(np.linspace(0, 10, 2000))]).T
df = dd.from_array(data, columns=['index', 'signal', 'id_group'], chunksize=100)
df1 = df.groupby("id_group").apply(dask_norm, meta=pd.Series(name="a") )
df2 = df1.to_frame().compute()
df3 = pd.concat([pd.DataFrame(a) for a in df2.a.values])
df3.columns = ["index", "signal", "id_group", "normalized_signal_by_group"]
df3.sort_values("index", inplace=True)

【讨论】：