问题描述
我非常熟悉 pandas 数据帧，但我对Dask太陌生，所以我仍在努力实现代码的并行化。 我已经用 pandas 和 pandas 得到了我想要的结果，所以我想弄清楚的是，我是否可以用Dask来扩大任务的规模或以某种方式加快它的速度。
假设我的DataFrame将DateTime作为非唯一索引、一个值列和一个id列。
time value id
2021-01-01 00:00:00.210281 28.08 293707
2021-01-01 00:00:00.279228 28.07 293708
2021-01-01 00:00:00.697341 28.08 293709
2021-01-01 00:00:00.941704 28.08 293710
2021-01-01 00:00:00.945422 28.07 293711
... ... ...
2021-01-01 23:59:59.288914 29.84 512665
2021-01-01 23:59:59.288914 29.83 512666
2021-01-01 23:59:59.288914 29.82 512667
2021-01-01 23:59:59.525227 29.84 512668
2021-01-01 23:59:59.784754 29.84 512669
我要提取的是每秒的最新值。例如，如果2021-01-01 00:00:01之前的价格是索引为2021-01-01 00:00:00.945422的行，则最新的值为28.07。
在我的例子中，有时索引值不是唯一的，因此为了打破平局，我想使用id列。具有最大id数字的值将被视为最新值。对于在时间2021-01-01 23:59:59.288914并列的三个值的情况，将选择29.82值，因为该日期的最大id将是512667。另请注意，id在整个数据集中并不一致，我不能仅依靠它来对我的数据进行排序。
在 pandas 身上，我只需获取最后一个索引即可
last_index = df.loc[date_minus60: date_curr].index[-1]
last_values = df.loc[last_index]
如果last_values.index.is_unique的值为FALSE，则最后执行last_values.sort_values('id').iloc[-1]。
我一直在将此代码转换为DASK时遇到困难，遇到了有关延迟函数的问题，导致它们需要计算才能再次重新索引我的数据帧。
我想知道是否有处理此类问题的最佳做法。
推荐答案
下面的代码片段显示它的语法非常相似：
import dask
# generate dask dataframe
ddf = dask.datasets.timeseries(freq="500ms", partition_freq="1h")
# generate a pandas dataframe
df = ddf.partitions[0].compute() # pandas df for example
# sample dates
date_minus60 = "2000-01-01 00:00:00.000"
date_curr = "2000-01-01 00:00:02.000"
# pandas code
last_index_pandas = df.loc[date_minus60:date_curr].index[-1]
last_values_pandas = df.loc[last_index_pandas]
# dask code
last_index_dask = ddf.loc[date_minus60:date_curr].compute().index[-1]
last_values_dask = ddf.loc[last_index_dask].compute()
# check equality of the results
print(last_values_pandas == last_values_dask)
dask版本中，区别在于两个.compute步骤，因为需要计算两个惰性值：第一个是找到正确的索引位置，第二个是获得实际值。此外，这还假设数据已经按时间戳编制了索引，如果没有，最好在加载到dask之前为数据编制索引，因为.set_index通常是一个较慢的操作。然而，根据您在此之后的实际用途，dask可能不是一个很好的用途。如果其基本思想是进行快速查找，那么更好的解决方案是使用索引数据库(包括专门的时间序列数据库)。
最后，上面的代码片段使用了唯一索引。如果实际数据具有非唯一索引，那么根据最大id选择的要求应该是在计算出last_values_dask之后处理的，方法如下(伪代码，不会立即生效)：
def get_largest_id(last_values):
return last_values.sort_values('id').tail(1)
last_values_dask = get_largest_id(last_values_dask)
如果查找的是批次(而不是特定的样本日期)，则可以设计更好的管道。
这篇关于提取具有非唯一索引列日期的DaskDataFrame中的最新值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！