迭代 DASK 延迟字典

Question

我有很多从 dask 延迟函数返回的延迟字典。我想将它们汇总成如下所示的summary_dict。items函数不适用于延迟对象。

@dask.delayed
def get_dict(date):
return {
'a': {'date': date},
'b': {'date': date}
}
summary_dict = {}
dates = [d1, d2, d3, ...]
for date in dates:
date_dict = get_dict(date)
# the following doesn't work because date_dict is a delayed object
for key, val in date_dict.items():
summary_dict.setdefault(key, []).append(val)

我能够执行以下操作来使其工作。但是，它非常丑陋，因为我必须对前面的密钥进行硬编码。

hardcoded_keys = ['a', 'b']
get_dict_items = {key, dask.delayed(operator.itemgetter(key)) for key in keys}
summary_dict = {}
dates = [d1, d2, d3, ...]
for date in dates:
date_dict = get_dict(date)
for hardcoded_key in hardcoded_keys:
val = get_dict_items[key](date_dict)
summary_dict.setdefault(key, []).append(val)

有没有更好的方法来实现这一目标？

Answer 1

我唯一可以添加到您的解决方案@abisko的是延迟对象支持__getitem__，因此您应该能够直接索引，而不是使用operator.itemgetter，您应该能够直接索引：

from dask import delayed
from datetime import datetime

@delayed
def get_dict(date):
return {
'a': {'date': date},
'b': {'date': date}
}

summary_dict = {}
dates = [datetime.now(), datetime(2020, 1, 1), datetime(1999, 4, 23)]
date_dicts = [get_dict(d) for d in dates]
for date_dict in date_dicts:
for key in ["a", "b"]:
summary_dict.setdefault(key, []).append(date_dict[key])

可能还值得注意的是，公共延迟 API 接受nout参数，这将支持遍历元组，但不幸的是不是字典。