昨天下午对HDFS的速度进行了测试，晚上又对Hama的peer间通信通信速度进行了测试。

[转载引用请注明出处：http://blog.csdn.net/bhq2010/article/details/8741647]

软硬件环境：

和之前的hdfs测试中用的是一样的：http://blog.csdn.net/bhq2010/article/details/8740154

hama安装的是0.6.0版本。

测试过程与结果：

在setup方法中选出一个master task作为主peer

在bsp中写了2个超步，第一个超步读取本地的文件，并将其一部分发送给master

master在第二个超步中接收到其他peer发过来的消息（数据），将其写入本地的文件中。

有一点需要注意，应该设定bspTask个数为集群中节点的个数，这样通常每个节点上会有且仅有一个bsp任务。少了会使得有些节点上没有bsp任务，多了会使得一个节点上的多个bsp任务同时读取一个文件，然后就挂掉了。

测试程序如下：

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.BSPJobClient;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.ClusterStatus;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.NullInputFormat;
import org.apache.hama.bsp.TextOutputFormat;
import org.apache.hama.bsp.sync.SyncException;

public class HamaTest
{
	private static Path TMP_OUTPUT = new Path("/tmp/pi-"
			+ System.currentTimeMillis());
	
	public static class CommunicationTest extends
			BSP<NullWritable, NullWritable, Text, Text, Text>
	{
		public static final Log LOG = LogFactory.getLog(CommunicationTest.class);
		private String masterTask;

		@Override
		public void bsp(
				BSPPeer<NullWritable, NullWritable, Text, Text, Text> peer)
				throws IOException, SyncException, InterruptedException
		{
			File f = new File("/data/external_links_en.nt");
			if (f.exists())
			{
				int i = 0;
				FileReader fr = new FileReader("/data/external_links_en.nt");
				BufferedReader reader = new BufferedReader(fr);
				String line = null;
				while ((line = reader.readLine()) != null)
				{
					i++;
					if (i > 661700)
					{
						break;
					}
					peer.send(masterTask, new Text(line));
				}
				reader.close();
			}
			peer.sync();
			if (peer.getPeerName().equals(masterTask))
			{
				Text received;
				FileWriter fw = new FileWriter("/data/tmpres");
				BufferedWriter writer = new BufferedWriter(fw);
				while ((received = peer.getCurrentMessage()) != null)
				{
					writer.write(received.toString() + "\n");
				}
				writer.close();
			}
			peer.sync();
		}

		@Override
		public void setup(
				BSPPeer<NullWritable, NullWritable, Text, Text, Text> peer)
				throws IOException
		{
			// Choose one as a master
			String[] allPeerNames = peer.getAllPeerNames();
			int port = 0;
			for (String peerName : allPeerNames)
			{
				if (peerName.split(":")[0].equals("iir455-200"))
				{
					if (port == 0
							|| Integer.parseInt(peerName.split(":")[1]) < port)
					{
						port = Integer.parseInt(peerName.split(":")[1]);
						masterTask = peerName;
					}
				}
			}
			try
			{
				peer.sync();
			} catch (SyncException e)
			{
				e.printStackTrace();
			} catch (InterruptedException e)
			{
				e.printStackTrace();
			}
		}

		@Override
		public void cleanup(
				BSPPeer<NullWritable, NullWritable, Text, Text, Text> peer)
				throws IOException
		{

		}
	}

	public static void main(String[] args) throws InterruptedException,
			IOException, ClassNotFoundException
	{
		HamaConfiguration conf = new HamaConfiguration();

		BSPJob bsp = new BSPJob(conf, HamaTest.class);
		bsp.setJobName("Connection Speed Test");
		bsp.setBspClass(CommunicationTest.class);
		bsp.setInputFormat(NullInputFormat.class);
		bsp.setOutputKeyClass(Text.class);
		bsp.setOutputValueClass(Text.class);
		bsp.setOutputFormat(TextOutputFormat.class);
		FileOutputFormat.setOutputPath(bsp, TMP_OUTPUT);

		BSPJobClient jobClient = new BSPJobClient(conf);
		ClusterStatus cluster = jobClient.getClusterStatus(true);

		if (args.length > 0)
		{
			bsp.setNumBspTask(Integer.parseInt(args[0]));
		} else
		{
			bsp.setNumBspTask(cluster.getMaxTasks());
		}

		long startTime = System.currentTimeMillis();
		if (bsp.waitForCompletion(true))
		{
			System.out.println("Job Finished in "
					+ (System.currentTimeMillis() - startTime) / 1000.0
					+ " seconds");
		}
	}
}

一共有6个节点上有数据，每个读取前661700行，大约70MB，六个一共410MB，做了5次，第1次和第4次报了java.io.IOException: java.lang.OutOfMemoryError: Java heap space错误，其他用时分别是：117.733秒、127.06秒、94.117秒。报内存不足也不出意料，如果再把数据量加大一点，那么就根本跑不完了，参考http://blog.csdn.net/bhq2010/article/details/8548070，默认配置下Hama是将要发送的和接收到的消息都缓存在内存中的，而主节点上剩余的内存只有3GB左右，运行Hama可能还要吃掉很大一部分。

如果改成只发前66170行，则用时在20秒左右，可见时间消耗主要在通信上。

小结：

Hama的peer之间通信速度和健壮性都不理想：

1、从六个节点向一个节点传410MB的消息居然平均用了110秒，去掉启动任务的大约10秒钟，其平均的传输速度只有4MB/s；

2、非常吃内存，剩余将近3GB的内存，竟然跑一个几百兆通信量的Job就会报内存不足，当然这也可能是Hama配置的问题，睡完觉了再查查文档；

所以还是不要用hama本身的同步通信功能传递大量的数据，它只适合在同步计算过程中发送少量的消息。