您好,登錄后才能下訂單哦!
這篇文章給大家分享的是有關Hadoop怎么實現HelloWorld的內容。小編覺得挺實用的,因此分享給大家做個參考,一起跟隨小編過來看看吧。
先從源碼看起,再一步步剖析
package org.apache.hadoop.examples; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.set("mapred.job.tracker", "172.16.10.15:9001");//自己額外加的代碼 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); } Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
大家可以看到整個源代碼分為三個部分:
1. Map
public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } }
a) 定義一個自己的Map過程,TokenizerMapper 這個類名自己設定,這個類需要繼承org.apache.hadoop.mapreduce包中的Mapper類,四個參數分別表示輸入鍵key的參數類型,輸入值value的參數類型,輸出鍵key的參數類型,輸出值value的參數類型。 值得注意的是Hadoop本身提供了一套可優化的網絡序列化傳輸的基本類型,而不是用java內嵌的類型。這些類型都是在org.apache.hadoop.io包中。其中LongWritable類型相當于Long類型,Text類型相當于String類型,IntWritable相當于Integer類型。
b) map方法中參數value是指文本文件中的一行,參數key是為該行首字母相對于文本文件首地址的偏移量
c) StringTokenizer類是一個用來分隔String的應用類,類似于split。
//它的構造函數有三種: public StringTokenizer(String str) public StringTokenizer(String str,String delim) public StringTokenizer(String str,String delim,boolean returnDelims) //其中第一個參數為要分隔的String,第二個參數為分隔字符集合,第三個參數為分隔符是否作為標記返回,如果不指定分隔符,默認是'\t\n\r\f' //它的方法主要有三種: public boolean hasMoreTokens()//返回是否還有分隔符 public String nextToken()//返回從當前位置到下一個分隔符的字符串 public int countTokens()//返回nextToken方法被調用的次數
d) 經過StringTolenizer 處理之后會得到一個個 < word,1 > 這樣的鍵值對,放在context里,Context用于輸出內容的寫入,讀起來有點兒繞口,自己理解一下。
2. Reduce
public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }
a) 同mapper 過程一樣,Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reducer類,并重寫其reduce方法。
b) reduce方法中輸入參數key 指單個單詞,values 指對應單詞的計數值的列表
c) reduce 方法的目的就是對列表的值進行加和處理
d) 輸出的是< key,value>,key 指單個單詞,value 指對應單詞的計數值的列表的值的總和。
3. Main
public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.set("mapred.job.tracker", "172.16.10.15:9001");//自己額外加的代碼 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); } Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }
a) Configuration conf = new Configuration(); 默認情況下,Configuration開始實例化的時候,會從Hadoop的配置文件里讀取參數。
b) conf.set(“mapred.job.tracker”, “172.16.10.15:9001”);設置這句代碼是由于我們要把使用eclipse提交作業到Hadoop集群,所以手動添加Job運行地址。若是直接在Hadoop 集群進行運行,不用加這句代碼。 而且你可以看到只要前三句使用了這個代碼,所以這三句以后的代碼才是所有Hadoop例子中都會包含的。
c) 接下來這一句也是讀取參數,這里是從命令行參數里讀取參數。
d) Job job = new Job(conf, “word count”); 在MapReduce處理過程中,由Job對象負責管理和運行一個計算任務,然后通過Job的若干方法來對任務的參數進行設置。”word count”是Job的名字,(當然了,根據所有java語言規范規定的那樣,你也可以用
Job job = new Job(); job.setJobName("Name");
的形式做聲明)。
e) job.setJarByClass(WordCount.class);是根據WordCount類的位置設置Jar文件 。
為什么要這么做?因為我們在Hadoop集群上運行這個作業時候,要把代碼打包成一個JAR文件,用以在集群上發布這個文件。Hadoop利用這個傳遞進去的類來查找包含它的JAR文件。
f) job.setMapperClass(TokenizerMapper.class);設置Mapper
g) job.setCombinerClass(IntSumReducer.class);設置Combiner,這里先使用Reduce類來進行Mapper 的中間結果的合并,能夠減輕網絡傳輸的壓力。
h) job.setReducerClass(IntSumReducer.class);設置Reduce
i) job.setOutputKeyClass(Text.class);和 job.setOutputValueClass(IntWritable.class);分別是設置輸出鍵的類型和設置輸出值的類型
j) FileInputFormat.addInputPath(job, new Path(otherArgs[0]));設置輸入文件,它是otherArgs第一個參數
k) FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));設置輸出文件,將輸出結果寫入這個文件里,它是otherArgs第二個參數 。
注意:在運行作業前這個輸出目錄不應該存在,否則Hadoop會報錯并拒絕運行該作業。這種預防措施的目的是防止數據丟失(如果長時間運行的數據結果被意外覆蓋,肯定是非常惱人的)。
l) System.exit(job.waitForCompletion(true) ? 0 : 1);job執行,等待執行結果
4. 各個包的功能
到此為止,三大部分就分析完畢,然后再來看看引入的有哪些類:
a) package org.apache.hadoop.examples;Java 提供包機制管理代碼,關鍵詞是package, 包名字可以自己定,但不能重復。通常為了包的唯一性,推薦使用公司域名的逆序作為包,于是有了上面例子中的‘org.apache.hadoop’這樣的包名。
b) import java.io.IOException; 凡是以java開頭的包,在JDK1.7的API里可以找到類的資料。這里是從java.io中引入IOException,是一個輸入輸出異常類。
c) import java.util.StringTokenizer;這是從java.util包中引入的StringTokenizer類,是一個解析文本的類。具體用法上文中已提過了。
d) import org.apache.hadoop.conf.Configuration;凡是以org.apache.hadoop開頭的包,在Hadoop1.2.1 的API文檔可以找到類的資料。這里是從hadoop的conf包中引入Configuration類,它是一個讀寫和保存配置信息的類。
e) import org.apache.hadoop.fs.Path; Path類保存文件或者目錄的路徑字符串
f) import org.apache.hadoop.io.IntWritable; IntWritable是一個以類表示的可序化的整數。在java中,要表示一個整數,可以使用int類型,也可以使用integer類型,integer封裝了int類型,且integer類是可序化的。但Hadoop認為integer的可序化不合適,于是實現了IntWritable。
g) import org.apache.hadoop.io.Text; 從io包中引入Text類,是一個存儲字符串的可比較可序化的類。
h) import org.apache.hadoop.mapreduce.Job; 引入Job類,Hadoop中每個需要執行的任務是一個Job,這個Job負責參數配置、設置MapReduce細節、提交到Hadoop集群、執行控制等操作。
i) import org.apache.hadoop.mapreduce.Mapper;引入Mapper類,負責MapReduce中的Map過程。
j) import org.apache.hadoop.mapreduce.Reducer;引入Reduce類,負責MapReduce中的Reduce過程。
k) import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;引入FileInputFormat類,主要功能是將文件進行切片。
l) import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;FileOutputFormat類是將輸出結果寫入文件。
m) import org.apache.hadoop.util.GenericOptionsParser;這個類負責解析命令行參數。
從代碼的功能上,我們已經對map reduce有了一個清晰的認識,那么wordcount程序具體是怎么執行的呢?
將文件file1.txt,file2.txt 上傳到hdfs中的hdfsinput1文件夾里(上傳的方式可以通過eclipse客戶端,也可以通過Hadoop命令行),然后在eclipse上編寫wordcount.java文件(也即是第一部分分析的源碼)
由于測試用的文件較小,所以每個文件為一個split,并將文件按行分割形成< key,value>,這一步由MapReduce框架自動完成,其中key值為該行首字母相對于文本文件首地址的偏移量。
得到map方法輸出的< key,value>對后,進行Combine操作。
同樣,在Reduce過程中先對輸入的數據進行排序,再交由自定義的reduce方法進行處理,得到新的< key,value>對,并作為WordCount的輸出結果,輸出結果存放在第一張圖的lxnoutputssss文件夾下的part-r-00000里。
感謝各位的閱讀!關于“Hadoop怎么實現HelloWorld”這篇文章就分享到這里了,希望以上內容可以對大家有一定的幫助,讓大家可以學到更多知識,如果覺得文章不錯,可以把它分享出去讓更多的人看到吧!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。