python dask dataframes - 将 groupby.apply 输出连接到单个数据帧

Question

我正在使用 dask dataframe.groupby().apply() 并获得一个 dask 系列作为返回值。 我是每个组的一个列表三元组，例如 (a,b,1)，然后希望将所有三元组变成一个单独的 dask 数据帧

我在映射函数的末尾使用此代码将三元组作为 dask df 返回

#assume here that trips is a generator for tripletes such as you would produce from itertools.product([l1,l2,l3])
trip = list(itertools.chain.from_iterable(trip))
df = pd.DataFrame.from_records(trip)
return dd.from_pandas(df,npartitions=1)

然后当我尝试使用类似于pandas concat 和dask concatenate 时

假设apply函数的结果是变量result。 我正在尝试使用 将 dask.dataframe 导入为 dd dd.concat(结果，轴=0

并得到错误

raise TypeError("dfs 必须是 DataFrames/Series 对象的列表") TypeError: dfs 必须是 DataFrames/Series 对象的列表

但是当我使用

检查结果类型时

print type(result)

我明白了

输出：类 'dask.dataframe.core.Series'

在一组 dask groupby 对象上应用函数并将所有结果放入一个数据帧的正确方法是什么？

谢谢

编辑：------------------------------ ----------------- 为了产生用例，假设这个假数据生成

import random
import pandas as pd
import dask.dataframe as dd
people = [[random.randint(1,3), random.randint(1,3), random.randint(1,3)] for i in range(1000)]
ddf = dd.from_pandas(pd.DataFrame.from_records(people, columns=["first name", "last name", "cars"]), npartitions=1)

现在我的任务是按名字和姓氏对人进行分组（例如，所有具有相同名字和名字的人），然后我需要获取一个新的 dask 数据框，其中将包含每个组有多少辆车。

假设 apply 函数可以返回一系列元组列表，例如 [(name,name,cars count),(name,name,cars count)] 或具有相同列的数据框 - name, name,车数。

是的，我知道可以通过其他方式解决特定用例，但请相信我，我的用例更复杂。但我不能共享数据，也不能生成任何类似的数据。所以让我们使用一个虚拟数据:-)

挑战是将应用的所有结果连接到单个 dask 数据帧中（pandas 数据帧将是一个问题，数据将不适合内存 - 因此通过 pandas 数据帧进行转换将是一个问题）

Answer 1

供您编辑：

In [8]: ddf.groupby(['first name', 'last name']).cars.count().compute()
Out[8]:
first name  last name
1           1            107
            2            107
            3            110
2           1            117
            2            120
            3             99
3           1            119
            2            103
            3            118
Name: cars, dtype: int64

Answer 2

如果apply 的输出是pandas DataFrame，对我来说工作，所以如果需要，最后转换为dask DataFrame：

def f(x):
    trip = ((1,2,x) for x in range(3))
    df = pd.DataFrame.from_records(trip)
    return df

df1 = ddf.groupby('cars').apply(f, meta={'x': 'i8', 'y': 'i8', 'z': 'i8'}).compute()
#only for remove MultiIndex
df1 = df1.reset_index()
print (df1)
   cars  level_1  x  y  z
0     1        0  1  2  0
1     1        1  1  2  1
2     1        2  1  2  2
3     2        0  1  2  0
4     2        1  1  2  1
5     2        2  1  2  2
6     3        0  1  2  0
7     3        1  1  2  1
8     3        2  1  2  2

ddf1 = dd.from_pandas(df1,npartitions=1)
print (ddf1)
                cars level_1      x      y      z
npartitions=1                                    
0              int64   int64  int64  int64  int64
8                ...     ...    ...    ...    ...
Dask Name: from_pandas, 1 tasks

编辑：

L = []
def f(x):
    trip = ((1,2,x) for x in range(3))
    #append each
    L.append(da.from_array(np.array(list(trip)), chunks=(1,3)))

ddf.groupby('cars').apply(f, meta={'x': 'i8', 'y': 'i8', 'z': 'i8'}).compute()
dar =  da.concatenate(L, axis=0)
print (dar)
dask.array<concatenate, shape=(12, 3), dtype=int32, chunksize=(1, 3)>