hama瀛︿範絎旇?錛?錛?peers涔嬮棿閫氫俏閫熷害嫻嬭瘯

hama瀛︿範絎旇?錛?錛?peers涔嬮棿閫氫俊閫熷害嫻嬭瘯

鏄ㄥぉ涓嬪崍瀵笻DFS鐨勯?搴﹁繘琛屼簡嫻嬭瘯錛屾櫄涓婂張瀵笻ama鐨刾eer闂撮?淇￠?淇￠?搴﹁繘琛屼簡嫻嬭瘯銆?/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鍦ㄧ?浜屼釜瓚呮?涓?帴鏀跺埌鍏朵粬peer鍙戣繃鏉ョ殑娑堟伅錛堟暟鎹?級錛屽皢鍏跺啓鍏ユ湰鍦扮殑鏂囦歡涓??

鏈変竴鐐歸渶瑕佹敞鎰忥紝搴旇?璁懼畾bspTask涓?暟涓洪泦緹や腑鑺傜偣鐨勪釜鏁幫紝杩欐牱閫氬父姣忎釜鑺傜偣涓婁細鏈変笖浠呮湁涓?釜bsp浠誨姟銆傚皯浜嗕細浣垮緱鏈変簺鑺傜偣涓婃病鏈塨sp浠誨姟錛屽?浜嗕細浣垮緱涓?釜鑺傜偣涓婄殑澶氫釜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錛岄粯璁ら厤緗?笅Hama鏄?皢瑕佸彂閫佺殑鍜屾帴鏀跺埌鐨勬秷鎭?兘緙撳瓨鍦ㄥ唴瀛樹腑鐨勶紝鑰屼富鑺傜偣涓婂墿浣欑殑鍐呭瓨鍙?湁3GB宸﹀彸錛岃繍琛孒ama鍙?兘杩樿?鍚冩帀寰堝ぇ涓?儴鍒嗐?

濡傛灉鏀規垚鍙?彂鍓?6170琛岋紝鍒欑敤鏃跺湪20縐掑乏鍙籌紝鍙??鏃墮棿娑堣?涓昏?鍦ㄩ?淇′笂銆?/p>灝忕粨錛?/h1>

Hama鐨刾eer涔嬮棿閫氫俊閫熷害鍜屽仴澹??閮戒笉鐞嗘兂錛?/p>

1銆佷粠鍏?釜鑺傜偣鍚戜竴涓?妭鐐逛紶410MB鐨勬秷鎭?眳鐒跺鈎鍧囩敤浜?10縐掞紝鍘繪帀鍚?姩浠誨姟鐨勫ぇ綰?0縐掗挓錛屽叾騫衝潎鐨勪紶杈撻?搴﹀彧鏈?MB/s錛?/p>

2銆侀潪甯稿悆鍐呭瓨錛屽墿浣欏皢杩?GB鐨勫唴瀛橈紝絝熺劧璺戜竴涓?嚑鐧懼厗閫氫俊閲忕殑Job灝變細鎶ュ唴瀛樹笉瓚籌紝褰撶劧杩欎篃鍙?兘鏄疕ama閰嶇疆鐨勯棶棰橈紝鐫″畬瑙変簡鍐嶆煡鏌ユ枃妗ｏ紱

鎵?互杩樻槸涓嶈?鐢╤ama鏈?韓鐨勫悓姝ラ?淇″姛鑳戒紶閫掑ぇ閲忕殑鏁版嵁錛屽畠鍙??鍚堝湪鍚屾?璁＄畻杩囩▼涓?彂閫佸皯閲忕殑娑堟伅銆?/p>