您将如何使用 Dask 在 Dask.Geodataframe 中递归查找相邻的多边形？

Question

我是 Dask 的新手。

我一直在努力让它完成以下任务：

我有两个地理数据框和一组：

# Main chunk and combined chunk are a list of polygons of tessellated cells

main_chunk = gpd.read_parquet(f"./out/singapore/tess_chunk_{int(n1)}.pq")
combined_chunks = main_chunk + adjacent chunks

# This is a set of uids in the main chunk
main_chunk_ids = set(main_chunk['uID'])

我一直在尝试通过 queen contiguity 分两个阶段将 main chunk 扩展到 3 order。首先，遍历主块中所有单元格的 uID，找到扩展块中的所有相邻单元格，如果它不在 main_chunk_ids 中则添加 uID，然后在每个相邻单元格上递归运行此函数，直到达到顺序 3

这是有效的非 Dask 版本：

# def neigh_look(cell, main_chunk_ids, order):
#     neighbours = combined_chunks[~combined_chunks.geometry.disjoint(cell.geometry)]
#     for index, neighbour in neighbours.iterrows():
        
#         if not neighbour["uID"] in main_chunk_ids:
#             main_chunk_ids.add(neighbour["uID"])

#              if order < 3:
#                 main_chunk_ids.union(neigh_look(neighbour, main_chunk_ids, order+1))
    
#     return main_chunk_ids

我一直在尝试对这段代码进行 dask-ify，但失败了；这是我到目前为止所拥有的，它使 python 崩溃了：

    %%time

queen_out = {}

def neigh_look(cell, main_chunk_ids, order):
    neighbours = combined_chunks_dask[~combined_chunks_dask.geometry.disjoint(cell.geometry)]
    for index, neighbour in neighbours.iterrows():
        
        if not neighbour["uID"] in main_chunk_ids:
            main_chunk_ids.add(neighbour["uID"])

        if order < 3:
           main_chunk_ids.union(neigh_look(neighbour, main_chunk_ids, order+1))
    
    gc.collect()

    return main_chunk_ids

for n1 in tqdm(range(1), total=1):
    main_chunk = gpd.read_parquet(f"./out/singapore/tess_chunk_{int(n1)}.pq")
    combined_chunks = main_chunk

    main_chunk_ids = set(main_chunk['uID'])
    queen_cells = main_chunk_ids

    for n2 in w.neighbors[n1]:
        neigh_chunk = gpd.read_parquet(f"./out/singapore/tess_chunk_{int(n2)}.pq")
        combined_chunks = combined_chunks.append(neigh_chunk)

    combined_chunks_dask = dgpd.from_geopandas(combined_chunks, npartitions=16)

    queen_area_delayed = []
    for index, row in main_chunk.iterrows():
        queen_area_delayed.append(delayed(neigh_look)(row, main_chunk_ids, 0))

        if index % 1000 == 0:
            gc.collect() # trigger garbage collection

    queen_area = dask.compute(*queen_area_delayed)
    queen_out[n1] = queen_area

任何帮助将不胜感激！

Answer 1

我通过完全放弃使用递归来修复它 -