在this post的基础上,我实现了自定义模式公式,但发现了该函数的性能问题。本质上,当我进入这个聚合时,我的集群只使用我的一个线程,这对性能不是很好。我正在对16k行中的150多个属性(主要是分类数据)进行计算,我认为我可以将这些属性分解为单独的线程/进程,稍后再重新组合到单个数据帧中。请注意,此聚合必须位于两列上,因此由于无法使用单个列作为索引,我的性能可能会变得更差
是否有办法将dask期货或并行处理合并到聚合计算中
import dask.dataframe as dd
from dask.distributed import Client
from pandas import DataFrame
def chunk(s):
return s.value_counts()
def agg(s):
s = s._selected_obj
return s.groupby(level=list(range(s.index.nlevels))).sum()
def finalize(s):
# s is a multi-index series of the form (group, value): count. First
# manually group on the group part of the index. The lambda will receive a
# sub-series with multi index. Next, drop the group part from the index.
# Finally, determine the index with the maximum value, i.e., the mode.
level = list(range(s.index.nlevels - 1))
return (
s.groupby(level=level)
.apply(lambda s: s.reset_index(level=level, drop=True).argmax())
)
def main() -> DataFrame:
client = Client('scheduler:8786')
ddf = dd.read_csv('/sample/data.csv')
custom_mode = dd.Aggregation('custom mode', chunk, agg, finalize)
result = ddf.groupby(['a','b']).agg(custom_mode).compute()
return result
附带说明,我正在使用Docker来使用daskdev/dask(2.18.1)Docker映像启动我的调度程序和工作人员
最后,我使用futures基本上并行化了每列的聚合。因为我有很多列,所以将每个聚合传递给它自己的工作线程可以节省我很多时间。感谢David的评论以及the article on parallel workloads from the dask documentation
相关问题 更多 >
编程相关推荐