使用 Dask 进行基于日期时间索引的切片

如何解决使用 Dask 进行基于日期时间索引的切片

我有两个数据框：links 有两个名为 onset 和 offset 的日期时间列，每一行都是一个事件。另一个数据帧称为 sensors，使用频率为 1m 的日期时间索引进行索引，并且有约 600 列，每列对应一个传感器 ID。本质上，对于每个 links 行，我想使用起始值和偏移值作为时间范围来将 sensors 中的相关行切片，通过获取平均值将它们跨行聚合，然后连接单行links 数据帧的平均传感器值水平。我已经通过遵循 Pandas 代码设法做到了这一点，它可以工作，但我有大量数据并且速度非常慢。

def search_sensors(sensors,start_time,stop_time):
    s = sensors[start_time: stop_time]
    s = s.mean()
    return s

# add column names of sensor-ids
links[sensors.columns] = None

for index,row in links.iterrows():
    start_time = row['start_time']
    stop_time = row['stop_time']
    mean_sensors = search_sensors(sensors,stop_time)
    links.iloc[index,sensors.columns] = mean_sensors.to_list()

我已经用 Dask 尝试了一些东西，但没有成功。

1. 在 Pandas 中使用 dask.delayed()，我得到一个 UserWarning: Large object of size 35.62 MiB detected in task graph:

    mean_sensors_list = []

    for index,row in links.iterrows():
        start_time = row['start_time']
        stop_time = row['stop_time']
        mean_sensors = dask.delayed(search_sensors)(sensors,stop_time)
        links_list.append(mean_sensors)   # mean_sensors is delayed object containing a pandas.Series of shape (600,nan) 

    results = dask.compute(*mean_sensors_list)

2. 在以下代码中使用 dask.dataframe() 与 Pandas 一样慢，而且我在 Dask 仪表板中看不到任何并行化指示。

    sensors_dd = dd.from_pandas(sensors_interp,npartitions=1)
    links_dd = dd.from_pandas(links,npartitions=1)

    mean_sensors_list = []

    for index,row in links_dd.iterrows():
        start_time = row['start_time']
        stop_time = row['stop_time']
        mean_sensors = search_sensors(sensors_dd,stop_time)
        links_list.append(mean_sensors)   # mean_sensors is a dask.Series of shape (600,nan)

    results = dask.compute(*mean_sensors_list)

3. 同时使用 1 和 2，即 mean_sensors = dask.delayed(search_sensors)(sensors_dd,stop_time)，mean_sensors 是包含 dask.Series 形状 (600,nan) 的延迟对象，但执行速度非常慢。仪表板显示了 3 个任务（search_sensors、finalize、from_pandas）的一些并行化，并且 4 个工作人员的 CPU 使用率非常低。另外，当我运行 Ubuntu 时，它会显示一条磁盘空间不足的消息。

解决方法

Dask 的新手，我不熟悉 map_partitions()。问题的解决方案如下：

res = links_dd.map_partitions(lambda df: df.apply((lambda row: search_sensors(sensors,row.start_time,row.stop_time)),axis=1)).compute()

在我的 4 核笔记本电脑中速度非常快。