如何使用Dask groupby修复“ValueError:所有键都必须是同一形状”

Question

我试图在一个groupby()上使用dask dataframe但是当增加分区数量时，收到标题中提到的值错误。groupby()应用于非索引列。在

实际的数据集是一个包含约14000000行的CSV。我尝试了一个140000行的小样本，对于npartition={2,3或4}，一切正常，但是设置npartition=5会导致值错误。
在整个CSV（1400万行）中阅读，并使用“dd.read_csv地址（）“并将npartition设置为任意值（dask自动选择npartition=40）也会导致值错误。在

我还尝试对非索引列“vendor_id”使用set_index（），它返回一个空的数据帧—我也不明白

我使用的是ubuntu18.04 fresh conda安装 达斯克1.2.0 熊猫0.24.2 数字1.16.3

示例代码：

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


#small sample df:
dict_sample= {'vendor_id': {0: 'CMT',  1: 'CMT',  2: 'CMT',  3: 'CMT',  4: 'VTS',  5: 'VTS',  6: 'VTS',  7: 'VTS',  8: 'CMT',  9: 'VTS',  10: 'VTS',  11: 'VTS',  12: 'VTS',  13: 'VTS'},
 'pickup_datetime': {0: '2009-01-22 11:21:35',  1: '2009-01-22 21:17:22',  2: '2009-01-09 22:25:13',  3: '2009-01-23 17:20:01',  4: '2009-01-24 23:18:00',  5: '2009-01-26 22:03:00',  6: '2009-01-02 15:58:00',  7: '2009-01-16 19:38:00',  8: '2009-01-13 18:57:06',  9: '2009-01-09 14:51:00',  10: '2009-01-14 18:15:00',  11: '2009-01-02 23:17:00',  12: '2009-01-31 09:59:00',  13: '2009-01-19 14:57:00'},
 'passenger_count': {0: 1,  1: 1,  2: 2,  3: 1,  4: 1,  5: 1,  6: 5,  7: 5,  8: 1,  9: 1,  10: 1,  11: 1,  12: 1,  13: 1}}


df_ = pd.DataFrame(dict_sample)


#dask:

ddf_raw = dd.from_pandas(df_, npartitions=3)

ddf_raw['pickup_datetime'] = dd.to_datetime(ddf_raw['pickup_datetime'])


#using groupby:
date_grouper = pd.Grouper(key='pickup_datetime',freq='1H',label="left")
ddf_raw_grouped = ddf_raw.groupby(['vendor_id',date_grouper]).passenger_count.count()

ddf_raw_grouped.head()

使用140000行和npartitions=5会导致：

^{pr2}$

正在尝试设置索引（）

ddf_raw = ddf_raw.set_index('vendor_id')
ddf_raw.head()

退货

Empty DataFrame
Columns: [pickup_datetime, passenger_count]
Index: []

我对dask比较陌生，我不知道是什么触发了ValueError。使用groupby（）在pandas中对140000行的样本进行操作没有任何问题。在

此外，我注意到玩npartitions也会改变“乘客的最终结果_计数。计数（）“值

最后，我尝试用另一个数据集复制错误

import dask
df_dask = dask.datasets.timeseries()

df_dask = df_dask.repartition(npartitions= 20)
f = pd.Grouper(key='timestamp',freq='1H',label="left")
g = pd.Grouper(key='name')
df_dask = df_dask.groupby([g, f]).x.count()
df_dask.head()

在这种情况下，groupby（）工作没有任何问题，但是“x.count（）”的最终结果仍然会随着不同的npartitions值而改变。在

我看不出我的CSV有什么问题

更新：

我已经成功地将groupby（）用于以下解决方法，但在另一列（“pickup_id”）上：

ddf_raw = ddf_raw.set_index("pickup_id")

ddf_raw = ddf_raw.persist()

def groupby_date_pickup(df,grouper_1,grouper_2):

    df_local = df[[grouper_1,grouper_2]].copy()

    date_grouper = pd.Grouper(key=grouper_1,freq='1H',label="left")
    df_local_grouped = df_local.groupby([df_local.index, date_grouper])[grouper_2].count()

    print("Grouping Step")
    return df_local_grouped


grouped_ddf = ddf_raw.map_partitions(groupby_date_pickup, "pickup_datetime", "passenger_count")