我正在用包装的C ++扩展在Python中构建分布式空间索引。我正在尝试使用Dask（第一次）而不是MPI。 当前的实现可行，但是存在一些问题。
-
关键部分是与串行执行相比，开销很大。我希望该算法几乎可以线性加速。
-
Persist似乎没有达到我的期望，因为第一个查询的时间比第二个查询的时间长。
-
在我看来，代码看起来很简单，但是正如我所说，Dask是我的新手。有更好的方法吗？
-
使用2d块时，我需要对延迟的对象进行一些索引。一维大块不会发生这种情况，感觉很奇怪。
该算法构建粒子的排序列表，然后构建八叉树，其中节点通过一对（开始，结束）索引在排序数组中通过引用粒子的连续块。查询采用边界框，并在八叉树中搜索重叠的节点，从所得候选中实际收集边界框中的粒子。构建和查询都是纯串行的。
并行化是通过随机拆分粒子，在每个子集上构建树来完成的。查询在所有子索引中广播，并且结果重新连接在一起。 Dask在这里感觉很自然。我使用persist一次生成索引并保持索引不变，因为我希望每个索引有很多查询。
我尝试了map_blocks，但这似乎仅适用于从数组到数组的转换。此外，还尝试了持久/计算的各种排列。
from dask.delayed import delayed
from dask.distributed import LocalCluster,Client
import dask.array as da
# My custom octree index
from pyoctree import Index
# For reference,this runs on a 10 Core x 8 Thread cluster (POWER8)
cluster = LocalCluster()
client = Client()
# Point count
N = 100_000_000
# Dimensionality
D = 3
# Chunk Count
K = 10
# Octree Levels
L = 5
# Delayed adapters for Dask
@delayed
def index(xs):
"""Create index"""
return Index(L,xs)
@delayed
def query(index,ll,ur):
"""Bounding box query between lower left and upper right"""
return index.query(*ll,*ur)
# Random input: N random points in 3d
array = da.random.random((N,3),chunks=(N//K,3)).astype(np.float32)
# Split array
splits = array.to_delayed()
# Create index
indices = [index(split[0]).persist() for split in splits]
# This takes roughly 100ms
# Execute a query
queries = [query(index,(0.25,0.25,0.25),(0.35,0.35,0.35)) for index in indices]
# Merge results
parts = [da.from_delayed(query,shape=(np.nan,dtype=np.float32) for query in queries]
parts = [part.compute_chunk_sizes() for part in parts]
final = da.concatenate(parts)
result = final.compute()
# This takes 100s the first time and 800ms afterwards
使用相同参数进行串行执行需要240秒钟来建立索引，而单个查询则需要6毫秒。使用Dask时，它的构建时间为100秒，每个查询的时间为800ms。将粒子数N增加10倍，每个查询得到11秒，构建索引得到240秒。因此，Dask似乎增加了严重的惩罚，以N为单位。
我很感谢任何提示。
最诚挚的问候