hama瀛板Q??peers涔俊害
ㄥぉ涓瀵DFS?搴琛涓瀵amaeer撮?淇￠?淇￠?搴琛?/p>
[杞插ㄨ敞澶http://blog.csdn.net/bhq2010/article/details/8741647]
杞纭浠剁澧hdfs涓ㄧ涓?http://blog.csdn.net/bhq2010/article/details/8740154
hama瀹0.6.0?/p>
etup涓?涓aster task浣涓eer
sp涓?涓瓒ワ涓涓瓒ヨ诲扮苟涓?master
masterㄧ浜釜ヤ腑ユ板浠eerョ娑锛Q跺ユ扮涓?
瑕敞搴ヨ之spTask涓颁负涓釜杩父姣釜涓浠涓?釜bsp浠浜浣垮涓病sp浠E浜浣垮涓?釜涓澶釜bsp浠璇涓?釜氨浜?
濡?/p>
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 extendsBSP<NullWritable, NullWritable, Text, Text, Text>{public static final Log LOG = LogFactory.getLog(CommunicationTest.class);private String masterTask;@Overridepublic 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();}@Overridepublic void setup(BSPPeer<NullWritable, NullWritable, Text, Text, Text> peer)throws IOException{// Choose one as a masterString[] 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();}}@Overridepublic 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");}}}涓??涓涓版锛涓璇?61700琛澶х害70MB涓涓?10MB浜?★?″?℃浜ava.io.IOException: java.lang.OutOfMemoryError: Java heap space浠跺?17.733?127.06?94.117?ュ瀛充涓@澶т?涔氨璺瀹?a href="http://blog.csdn.net/bhq2010/article/details/8548070">http://blog.csdn.net/bhq2010/article/details/8548070璁ら涓ama灏跺骄~ㄥ瀛腑富涓浣3GB宸琛ama借瑕澶т??
濡?6170琛跺20乏瑗娑?涓昏?淇′?/p>?/h1>
Hamaeer涔俊害澹?戒?/p>
1涓涓浼410MB灞邂h浜?10ㄤ徽dぇ?0杈?搴?MB/s?/p>
2甯稿浣杩?GB瀛璺涓俊Jobュ瀛褰杩芥Hama疆棰″瑙ユ妗ｏ
?互杩涓hama桧n姝ラ?淇″戒ぇ版?ヨ＄杩涓娑?/p>