大规模DAG多Walker最短路径并行计算：Dask与graph-tool协同难题

我有一个大型有向无环图（DAG），包含约1000万条边和10万个节点，同时有3万个Walker的列表。每个Walker对应源节点和目标节点，需要额外计算生成约50条边接入DAG，之后要为每个Walker高效计算最短路径，还要实现可扩展的并行处理。

我只熟练掌握Python，最初用Dask做并行化，生成额外边时效果还行，但计算最短路径必须用graph-tool针对DAG优化的快速C++实现。

现在不清楚怎么在Dask Worker间高效共享这个大型基础DAG，也不知道怎么给每个Walker映射添加节点/边并计算最短路径的函数。目前用的是8线程Linux虚拟机，内存足够支撑各线程计算，也希望能扩展到多台虚拟机提速。

原循环实现代码：

import graph_tool as gt

# 初始化graph-tool图
g = gt.Graph()

# 添加带权重的边
g.ep['weight'] = g.new_edge_property('double')
g.add_edge_list(edges[['src','dst','weight']].values, eprops=[g.ep['weight']])

for walker in walkers:
    origin_node       = g.add_vertex()
    destination_node  = g.add_vertex() 
    origin_edges      = function_to_compute_origin_edges(walker)
    destination_edges = function_to_compute_destination_edges(walker)
    g.add_edge_list([origin_edges,destination_edges])

    vertex_list,edge_list = gt.topology.shortest_path(g, origin_node, destination_node, weights=g.ep['weight'], dag=True)

    g.remove_vertices([origin_node,destination_node])

因为每个计算任务相互独立，希望把这个循环并行化处理。

1. 高效共享基础DAG：序列化+Worker预加载 #

graph-tool图无法跨进程直接共享，需先序列化到磁盘，让每个Dask Worker启动时加载到本地内存：

• 先序列化基础DAG（包含所有属性）：

  gt.save(g, "base_dag.gt", fmt="gt")
  

• 定义Worker预加载函数，让每个Worker启动时加载图：

  def load_base_graph():
      import graph_tool as gt
      global base_g
      base_g = gt.load_graph("base_dag.gt")
      base_g.ep['weight'] = base_g.edge_properties['weight']
  

• 通过Dask Client注册回调，确保Worker启动时执行加载：

  from dask.distributed import Client
  client = Client()
  client.register_worker_callbacks(setup=load_base_graph)
  

  这样每个Worker内存中都有一份独立的基础DAG副本，避免重复传输大文件。

2. 单个Walker的并行处理函数 #

每个任务需操作独立的图副本，避免修改共享的基础DAG：

def process_walker(walker):
    import graph_tool as gt
    # 复制基础图，生成独立副本
    g = base_g.copy()
    origin_node = g.add_vertex()
    destination_node = g.add_vertex()
    
    # 计算并添加walker专属边
    origin_edges = function_to_compute_origin_edges(walker)
    destination_edges = function_to_compute_destination_edges(walker)
    g.add_edge_list(origin_edges + destination_edges)
    
    # 计算DAG最短路径
    vertex_list, edge_list = gt.topology.shortest_path(g, origin_node, destination_node, weights=g.ep['weight'], dag=True)
    
    # 返回结果，临时图会自动回收
    return walker.id, vertex_list, edge_list

3. 批量提交并行任务 #

用Dask的client.map批量处理所有Walker：

# 提交所有任务
results = client.map(process_walker, walkers)
# 收集所有结果
final_results = client.gather(results)

如果Walker数量过多，可调整Dask的batch_size参数，平衡调度开销和并行度。

4. 多机器扩展要点 #

• 共享存储：确保所有Worker能访问base_dag.gt，用NFS、S3挂载或者本地复制文件。
• 资源配置：根据单个任务的内存占用（基础DAG+临时边）设置Worker的memory_limit，避免内存溢出。
• 集群管理：用dask-jobqueue或dask-kubernetes快速部署多机器Worker集群。

5. 性能优化 #

• 预计算Walker边：如果边计算逻辑独立，可先并行计算所有Walker的边，再提交路径计算任务，拆分计算步骤。
• 精简图属性：去掉基础DAG中不需要的属性，减少复制时的内存开销。
• 调整graph-tool参数：设置gt.openmp_set_num_threads(1)避免多线程冲突，因为Dask已经做了进程级并行。

内容的提问来源于stack exchange，提问作者Alessandro Zanoli