如何将 Dask DataFrame 与 Numpy Busday_count 一起使用？答案

【问题标题】：How to use Dask DataFrame with Numpy Busday_count?如何将 Dask DataFrame 与 Numpy Busday_count 一起使用？
【发布时间】：2020-09-21 21:15:54
【问题描述】：

我正在努力将 Pandas/Numpy 代码转换为 Dask 以处理更大的数据集。我似乎无法重新创建以下 Pandas/Numpy 代码：

df['days_to_complete'] = np.busday_count(begindates=df['time_order_date'].values.astype('datetime64[D]'),enddates=df['time_complete_date'],weekmask='1111111',holidays=hols_list)

在考虑工作周和假期列表时，这将返回 time_order_date 和 time_complete_date 之间的整数天数。它在我的数据框中创建并填充了一个新列，没有问题。

在 Dask 中，我尝试了以下方法：

map_partitions 调用 numpy 函数：

ddf['days_to_complete'] = ddf.time_order.map_partitions(func=np.busday_count,args= ddf['time_order_date'].values.astype('datetime64[D]'),ddf['time_complete_date']),meta=(None, 'i8'))
还有使用 lambda 的 map_partitions：

ddf['days_to_complete'] = ddf.map_partitions(lambda ddf: ddf.assign(result = np.busday_count(begindates=ddf['time_order_date'].values.astype('datetime64[D]'),enddates=ddf['time_complete_date'],weekmask='1111111',holidays=hols_list)),meta=(None,'i8'))

运行ddf.compute()后出现如下错误：

TypeError: busday_count() got multiple values for argument 'begindates'

如何以并行处理/对 Dask 友好的方式最好地使用这个 numpy 函数？ 我没有成功使用 Dask 文档/示例或其他 SO 线程。我还想使用 Pandas CustomBusinessHour rollfoward，就像我在这里使用基本的 pandas 一样：

bis_hour = CustomBusinessHour(n=1,weekmask='Mon Tue Wed Thu Fri Sat Sun',holidays=hols_list,start = bus_hours_start,end = bus_hours_end,offset=0)
df['time_order_bis'] = pd.to_datetime(df['time_order'])
df['time_order_bis'] = df['time_order_bis'].apply(lambda row: bis_hour.rollforward(row))

这会将订单时间“前滚”到定义的客户营业时间内（周六订单现在是周一早上 7 点，一个工作日）。谢谢！

编辑： 我试过写和调用一个函数：

def bdays(df):
  return np.busday_count(df.time_order_date.values.astype('datetime64[D]'),df.time_complete_date,weekmask='1111111',holidays=hols_list)
ddf['days_to_complete'] = ddf.map_partitions(bdays,df=ddf,meta=('days_to_complete','i8')).compute()

我收到以下错误：TypeError: bdays() got multiple values for argument 'df'

【问题讨论】：

标签： python pandas numpy dask dask-dataframe

【解决方案1】：

我搞定了！关键是返回一个 Dask 数组并且不要太快计算东西，这会破坏类型。我建议做大量的 type() 检查并一步一步地进行，本质上你需要 Dask 对象，pandas 对象/numpy 数组可以破坏分区/并行性。

功能：

def bdays(df=ddf):


return da.from_array(np.busday_count(df.time_order_date,df.time_complete_date,weekmask='1111111',holidays=hols_list))

使用 map_partitions。请注意上面函数的第一个参数需要数据框/分区 -> 我们不在地图分区中指定这个！只有附加参数。

ddf['days_to_complete'] = ddf.map_partitions(bdays,meta=('days_to_complete','i8'))

在分配给我的数据框中的新列之前计算 (compute()) 导致错误。

TypeError: set_index() missing 1 required positional argument: 'other'

调试建议：测试您的输入并仅使用一个分区测试功能。 bdays 是上面的函数。

type(ddf.map_partitions(bdays,meta='i8'))

output: dask.dataframe.core.Series

【讨论】：