用MPI实现Hadoop Map/Reduce的TeraSort

http://emonkey.blog.sohu.com/166546157.html

参与排序的MPI进程首先对输入文件进行划分，以确定每个进程计算的文件偏移量和长度，然后通过MPI-IO读取本进程需要排序的部分
每个进程对要处理的数据用qsort进行本地排序，然后收集本进程的最小键值和最大键值，min_key, max_key，通过MPI_Allreduce()调用，每个进程获得全局的最小和最大键值,g_min_key, g_max_key
通过g_min_key和g_max_key，将键值区间[g_min_key, g_max_key]按某种规则划分成N份，其中N是MPI进程总数，生成多个区间，[g_min_key, key1), [key1, key2), ...,[keyN, g_max_key+1), 每个MPI进程按本进程的rank号收集并排序对应区间内的记录。过程如下，每个MPI进程会统计本地的记录在每个区间的分布，通过MPI_Allgather，这样每个进程都会得到一个全局的键值分布图，通过该信息，MPI进程都清楚自己要从哪些MPI进程接收多少数据，以及要向其他哪些节点发送哪些数据。通过MPI_Irecv,
和MPI_Irsend实现数据的收集和分发。
每个进程对收集到的数据，再进行排序，此时可以直接排序，或是用merge排序，此时每个进程内的数据都是有序的，而且整个数据按进程的rank顺序是自然排好的
通过MPI_File_write_ordered将每个进程内排序好的数据写到输出文件

MPI实现海量数据分布式全局排序

83032057个记录，每个记录长度9840字节，数据总大小760.92GB, 分布存储在16个计算节点上，

每个节点有24GB内存，两个硬盘: 560GB /disk1是SCSI (xfs)，480GB /disk2是SATA (ext3), 节点间是千兆以太网，每个节点捆绑使用两个网卡。

记录数据放在/disk1, /disk2用做临时盘，排序好的数据放在/diks1,

节点间通信使用MPI, 排序算法内部分成两步部分，

第一部分是数据分发收集，0-15号节点所用时间如下：

0: time used 761.051 seconds

1: time used 761.06 seconds

2: time used 761.064 seconds

3: time used 761.065 seconds

4: time used 761.055 seconds

5: time used 761.216 seconds

6: time used 761.054 seconds

7: time used 761.054 seconds

8: time used 761.051 seconds

9: time used 761.063 seconds

10:time used 761.046 seconds

11:time used 761.066 seconds

12:time used 761.055 seconds

13:time used 761.083 seconds

14:time used 761.063 seconds

15:time used 761.122 seconds

第二部分是节点本地排序，0-15号节点所用时间如下：

0: time used 385.87 seconds

1: time used 407.582 seconds

2: time used 447.772 seconds

3: time used 416.993 seconds

4: time used 448.436 seconds

5: time used 446.271 seconds

6: time used 399.21 seconds

7: time used 549.768 seconds

8: time used 428.943 seconds

9: time used 430.585 seconds

10:time used 394.615 seconds

11:time used 505.52 seconds

12:time used 501.791 seconds

13:time used 392.03 seconds

14:time used 558.948 seconds

15:time used 496.497 seconds

总体上用1300秒，约22分钟，可以对亿级记录数据进行分布式全局排序！

可以完胜HADOOP！

采用多线程、异步IO、非阻塞MPI传输等技术！

体验新版博客