Open MPI与Dask结合使用mpirun执行报错求助

你的核心问题是每个MPI进程都独立启动了Dask Client，这会导致三个关键问题：

• 同时启动多个Dask集群，造成端口冲突、系统资源耗尽（这就是mpirun -np 3执行时出错的直接原因）
• 每个进程都完整计算了(x+x.T).sum()的结果，最后用MPI SUM会得到3倍的正确值
• Dask本身已是分布式计算框架，这种重复的MPI聚合逻辑完全冗余

下面给出两种可行的修改方案：

方案1：移除冗余MPI逻辑，纯用Dask分布式计算 #

如果你的需求只是分布式处理数组任务，不需要MPI的特定功能，直接用Dask的分布式调度即可：

from dask.distributed import Client
import dask.array as da

def main():
    # 启动Dask Client，自动创建本地集群（默认使用所有CPU核心）
    # 也可手动指定worker数量：Client(n_workers=3)
    client = Client()

    # 创建Dask数组并执行计算
    shape = (10000, 10000)
    chunks = (1000, 1000)
    x = da.ones(shape, chunks=chunks)
    result = (x + x.T).sum()

    # 获取结果并打印
    total = result.compute()
    print(f"Result: {total}")

if __name__ == "__main__":
    main()

运行方式：直接执行python test.py即可，Dask会自动调度任务到多个worker节点。

方案2：正确结合MPI与Dask（仅当必须依赖MPI时） #

如果业务逻辑确实需要MPI的进程间通信，正确的做法是仅在rank 0节点启动Dask Client，其余MPI进程作为Dask Worker连接到该Client，避免重复启动集群：

from mpi4py import MPI
from dask.distributed import Client, Worker
import dask.array as da

def main():
    comm = MPI.COMM_WORLD
    rank = comm.Get_rank()

    if rank == 0:
        # Rank 0启动Dask调度器与Client，暂不自动创建worker
        client = Client(n_workers=0)
        scheduler_address = client.scheduler_info()['address']
        # 将调度器地址广播给其他MPI进程
        comm.bcast(scheduler_address, root=0)

        # 定义计算任务并执行
        shape = (10000, 10000)
        chunks = (1000, 1000)
        x = da.ones(shape, chunks=chunks)
        result = (x + x.T).sum()
        total = result.compute()
        print(f"Result: {total}")

        # 清理资源
        client.close()
    else:
        # 其他MPI进程作为Dask Worker连接到调度器
        scheduler_address = comm.bcast(None, root=0)
        worker = Worker(scheduler_address)
        worker.run()

if __name__ == "__main__":
    main()

运行方式：mpirun -np 3 python test.py，此时rank0作为调度器+Client，另外2个进程作为Dask Worker执行计算任务。

额外注意事项 #

• VS Code直接运行时仅启动1个进程，不会出现多Client冲突问题，因此能正常执行
• 确保Open MPI与mpi4py版本兼容，Dask版本适配当前Ubuntu环境
• 无特殊MPI依赖时优先选择方案1，Dask的分布式调度已足够处理绝大多数数组计算场景

内容的提问来源于stack exchange，提问作者KIRAN TS