Python中的分布式并行编程：MPI4PY

1介绍

MPI代表消息传递接口。 MPI“或OpenMPI”等实现方式，用于创建平台，在分布式系统中编写并行程序，例如具有分布式存储器的Linux集群，通常使用MPI标准编写C平台，所以为了在Python环境中运行并行程序，我们需要使用一个名为MPI4py的模块，这意味着“MPI for Python”，该模块提供了标准的功能来执行处理器的排名，发送和接收消息/数据集群中的各种节点，允许程序与节点之间传递的消息并行执行，重要的是MPIch2和MPI4py安装在系统中，所以如果没有安装MPI4Py，以下是2个指南用于在MPI4PY中安装，构建和测试示例程序。

https://seethesource.wordpress.com/2015/01/05/raspberypi-hacks-part1/
https://seethesource.wordpress.com/2015/01/14/raspberypi-hacks-part2/

一旦安装了MPI4PY，就可以开始编程了。 本教程介绍MPI4PY提供的各种重要功能，如发送接收消息， 散射和收集数据以及广播消息，以及如何通过提供示例来使用。 使用这些信息，可以在Python中构建可扩展的高效分布式并行程序。 所以，让我们开始吧。

2发送和接收消息

mpi4py中的通信是使用send()和he recv()方法完成的。 顾名思义，它分别用于从节点发送和接收消息。

2.1发送简介()

这个函数的一般语法是： comm.send（data，dest）

这里的“数据”可以是必须发送到另一个节点的任何数据/消息，“dest”表示节点发送给它的进程等级。

示例：comm.send（（rank + 1）* 5，dest = 1）。
这将消息“（rank + 1）* 5”发送到process rank = 1的节点。 所以只有该节点可以接收它。

2.2 recv()

这个函数的一般语法是： comm.recv（source）

这告诉一个特定的过程，只能从“source”参数中提到的等级的进程接收数据/消息。

示例：comm.recv（source = 1）
这只接收来自rank = 1的进程的消息。

2.3简单的send()和recv()的例子

if rank==0 :
	    data= (rank+1)*5
	    comm.send(data,dest=1)
if rank==1:
	    data=comm.recv(source-0)
	    print data

（对于完整的实现程序，请参考Example1.py）

[ 下载Example1.py ]

2.4备注

• 当节点运行recv()方法时，它等待它从预期的源接收到一些数据。 一旦收到一些数据，它将继续与程序的其余部分。
• 这里，recv()中send()和“source”参数中的“dest”参数不需要只有一个常量值（或等级），它可以是一个表达式。
• “comm”对象的“size”成员是发送()和receive()方法的一个很好的方法，这使我们有动态的发送和接收消息。

2.5动态发送和接收

数据的动态传输更加有用，因为它允许同时由多个节点发送和接收数据，并且可以根据特定情况进行决定传输，从而大大提高了灵活性。

2.6动态发送和接收数据的示例

comm.send(data_shared,dest=(rank*2)%size)
comm.recv(source=(rank-3)%size)

上述两个语句是动态的，因为要发送的数据以及必须发送的数据取决于由等级和大小替代的值，这些值是动态发生的，因此不需要对值进行硬编码。 然而，recv()方法只接收一个消息，即使它有资格接收其中的许多消息，因此只接收它的第一个消息，它服务并继续到程序中的下一个语句。

（完全实现参考Example2.py）

[ 下载Example2.py ]

3标记为send()和recv()函数

当我们标记send()和recv()时，我们可以保证接收消息的顺序，因此我们可以确保在另一个消息之前传递一个消息

在数据动态传输过程中，出现这种情况，我们需要一个特定的send()来匹配一个特定的recv()来实现一种同步。 这可以使用send()和recv()中的“tag”参数完成。

例如，send()可以如下所示： comm.send（shared_data，dest = 2，tag = 1）和上述语句的匹配recv()将如下所示： comm.recv（source = 1，tag = 1）

因此，这种结构强制匹配，导致数据传输的同步。 标记的优点是可以使recv()等待，直到它从具有预期标签的相应send()接收数据。 但是，这必须非常小心地使用，因为它可能导致死锁状态。

3.1示例

  If rank==0:
	shared_data1 = 23
	comm.send(shared_data1,dest=3,tag=1)
	shared_data2 = 34
	comm.send(shared_data2,dest=3,tag=2)
  if rank==3:
	recv_data1 = comm.recv(source=0,tag=2)
	print  recv_data1
	recv_data2 = comm.recv(source=0,tag=1)
	print  recv_data2

输出结果如下所示：

34
23

因此，我们可以看到，即使shared_data1首先发送，第一个recv()等待tag = 2的send()并接收数据，打印并转发到下一个recv()方法。

（对于完整的实现，请参考Example3.py）

[ 下载示例3.py ]

4广播

广播允许用户将动态属性添加到并行编程，其中由主机生成的一些数据可以广播到所有节点。 这避免了向每个节点发送数据的重复过程。 从主节点/根节点，我们从该节点创建数据并将其发送到所有其他节点。

4.1示例

   if rank == 0
       data = {'a':1,'b':2,'c':3}
   else
       data = None
   data = comm.bcast(data,root=0)
   print  "rank", rank, data

在上述程序中，bcast()方法，第一个参数“data”表示必须广播的内容，第二个参数“root = 0”表示从哪里获取数据。如果我们使用5个进程运行此程序，输出应该类似于：

rank 0 {'a':1,'b':2,'c':3}
rank 4 {'a':1,'b':2,'c':3}
rank 3 {'a':1,'b':2,'c':3}
rank 1 {'a':1,'b':2,'c':3}
rank 2 {'a':1,'b':2,'c':3}

5散点

分散是分解数据并将每个部分发送或分发到不同节点的过程。一个例子是分解列表，并将每个列表元素发送到不同的节点。

5.1示例

   comm = MPI.COMM_WORLD
   size=comm.get_size()
   rank-comm.get_rank()
   if rank == 0: 
       data = [(x+1) ** x for x in range (size)]
       print 'scattering data',data
   else:
       data = None
   data = comm.scatter(data,root=0)
   print 'rank',rank,'has data: ', data

该程序的输出为：

scattering data : [1,2,9,64,625]

rank 0 has data : 1
rank 1 has data : 2
rank 2 has data : 9
rank 3 has data : 64
rank 4 has data : 625

这里的数据大小应该等于可以预期的数据数，也就是说如果数据有10个元素，而且只有5个进程存在，那么就会发生错误。

6聚集

收集与散点相反。 它用于从各种节点收集数据并将其存储为一个。一个例子是从各种通讯中收集元素，并从中排除单个列表。

6.1示例

示例：（请注意，与上一个散射示例相同，但有几个添加）

 comm = MPI.COMM_WORLD
   size=comm.get_size()
   rank-comm.get_rank()
   if rank == 0: 
       data = [(x+1) ** x for x in range (size)]
       print 'scattering data',data
   else:
       data = None
   data = comm.scatter(data,root=0)
   print 'rank',rank,'has data: ', data
     
   new_data = comm.gather(data, root=0)
   if rank == 0:
       print  'master collected: ', new_data
   

该程序的输出为：

rank 0 has data : 1
rank 1 has data : 2
rank 2 has data : 9
rank 3 has data : 64
rank 4 has data : 625 
master collected:  [ 1, 2, 3, 9,64, 625]  

在上面的输出中，最后一行是因为收集。 可以做的另一件事就是在分散数据时立即更改数据，即每个节点接收到一部分分散数据。 示例： data = data + 1
因此，当收集发生时，数据与发送的数据不同，但保持更改的模式，从而证明其正常工作

7结论

这篇文章介绍和使用了mpi4py的各种编程结构，它将帮助您使用python编程语言为分布式环境编写并行程序。 虽然这些是大多数程序中显示的单个构造，但全部或大部分都是一起使用的。 我想分享一些可以使用mpi4py构建的应用程序。

• 自动文本摘要 - 自然语言处理的一个分支，其中可以编写分布式程序来一次总结几个文本文档
• 排序和搜索 - 分布式环境可以通过分割要进行排序或搜索的数据进行使用，以便在少量时间内处理大量数据。
• 数学问题解决方案 （如毒药的分布）在并行环境中可以更快地解决

所以，我们可以为任何领域编写并行程序，并使用python提供的好处。