云智慧（北京）科技有限公司陳鑫

公司主營業(yè)務(wù)：做網(wǎng)站、網(wǎng)站設(shè)計(jì)、移動(dòng)網(wǎng)站開發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競爭能力。創(chuàng)新互聯(lián)建站是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊(duì)。公司秉承以“開放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對我們的高要求，感謝他們從不同領(lǐng)域給我們帶來的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會(huì)用頭腦與智慧不斷的給客戶帶來驚喜。創(chuàng)新互聯(lián)建站推出上高免費(fèi)做網(wǎng)站回饋大家。

 

寫這個(gè)文章的時(shí)候才意識(shí)到新舊API是同時(shí)存在于1.1.2的hadoop中的。以前還一直納悶兒為什么有時(shí)候是jobClient提交任務(wù)，有時(shí)是Job...不管API是否更新，下面這些類也還是存在于API中的，經(jīng)過自己跟蹤源碼，發(fā)現(xiàn)原理還是這些。只不過進(jìn)行了重新組織，進(jìn)行了一些封裝，使得擴(kuò)展性更好。所以還是把這些東西從記事本貼進(jìn)來吧。

關(guān)于這些類的介紹以及使用，有的是在自己debug中看到的，多數(shù)為純翻譯API的注釋，但是翻譯的過程受益良多。

GenericOptionsParser

parseGeneralOptions(Optionsopts, Configuration conf, String[] args)解析命令行參數(shù)
     GenericOptionsParser是為hadoop框架解析命令行參數(shù)的工具類。它能夠辨認(rèn)標(biāo)準(zhǔn)的命令行參數(shù)，使app能夠輕松指定namenode，jobtracker，以及額外的配置資源或信息等。它支持的功能有：
      -conf 指定配置文件；
      -D 指定配置信息；
      -fs      指定namenode
      -jt    指定jobtracker
      -files  指定需要copy到MR集群的文件，以逗號(hào)分隔
      -libjars指定需要copy到MR集群的classpath的jar包，以逗號(hào)分隔
      -archives指定需要copy到MR集群的壓縮文件，以逗號(hào)分隔，會(huì)自動(dòng)解壓縮

1.String[] otherArgs = new GenericOptionsParser(job, args)  

2.     .getRemainingArgs();  

3.if (otherArgs.length != 2) {  

4.   System.err.println("Usage: wordcount <in> <out>");  

5.   System.exit(2);  

6.}  

ToolRunner

用來跑實(shí)現(xiàn)Tool接口的工具。它與GenericOptionsParser合作來解析命令行參數(shù)，只在此次運(yùn)行中更改configuration的參數(shù)。
Tool

處理命令行參數(shù)的接口。Tool是MR的任何tool/app的標(biāo)準(zhǔn)。這些實(shí)現(xiàn)應(yīng)該代理對標(biāo)準(zhǔn)命令行參數(shù)的處理。下面是典型實(shí)現(xiàn)：

1.public class MyApp extends Configured implements Tool {  

2.      

3.   public int run(String[] args) throws Exception {  

4.     // 即將被ToolRunner執(zhí)行的Configuration  

5.     Configuration conf = getConf();  

6.       

7.     // 使用conf建立JobConf  

8.     JobConf job = new JobConf(conf, MyApp.class);  

9.       

10.     // 執(zhí)行客戶端參數(shù)  

11.     Path in = new Path(args[1]);  

12.     Path out = new Path(args[2]);  

13.       

14.     // 指定job相關(guān)的參數(shù)       

15.     job.setJobName("my-app");  

16.     job.setInputPath(in);  

17.     job.setOutputPath(out);  

18.     job.setMapperClass(MyApp.MyMapper.class);  

19.     job.setReducerClass(MyApp.MyReducer.class);  

20.*  

21.     // 提交job，然后監(jiān)視進(jìn)度直到j(luò)ob完成  

22.     JobClient.runJob(job);  

23.   }  

24.     

25.   public static void main(String[] args) throws Exception {  

26.     // 讓ToolRunner 處理命令行參數(shù)   

27.     int res = ToolRunner.run(new Configuration(), new Sort(), args);  //這里封裝了GenericOptionsParser解析args  

28.       

29.     System.exit(res);  

30.   }  

31. }  

MultipleOutputFormat 
自定義輸出文件名稱或者說名稱格式。在jobconf中setOutputFormat(MultipleOutputFormat的子類)就行了。而不是那種part-r-00000啥的了。。。并且可以分配結(jié)果到多個(gè)文件中。
      MultipleOutputFormat繼承了FileOutputFormat,允許將輸出數(shù)據(jù)寫進(jìn)不同的輸出文件中。有三種應(yīng)用場景：

a.最少有一個(gè)reducer的mapreduce任務(wù)。這個(gè)reducer想要根據(jù)實(shí)際的key將輸出寫進(jìn)不同的文件中。假設(shè)一個(gè)key編碼了實(shí)際的key和為實(shí)際的key指定的位置

b.只有map的任務(wù)。這個(gè)任務(wù)想要把輸入文件或者輸入內(nèi)容的部分名稱設(shè)為輸出文件名。

c.只有map的任務(wù)。這個(gè)任務(wù)為輸出命名時(shí)，需要依賴keys和輸入文件名。 

1.//這里是根據(jù)key生成多個(gè)文件的地方，可以看到還有value，name等參數(shù)  

2.@Override  

3.protected String generateFileNameForKeyValue(Text key,  

4.     IntWritable value, String name) {  

5.   char c = key.toString().toLowerCase().charAt(0);  

6.   if (c >= 'a' && c <= 'z') {  

7.     return c + ".txt";  

8.   }  

9.   return "result.txt";  

10.}  

 

DistributedCache

在集群中快速分發(fā)大的只讀文件。DistributedCache是MR用來緩存app需要的諸如text，archive，jar等的文件的。app通過jobconf中的url來指定需要緩存的文件。它會(huì)假定指定的這個(gè)文件已經(jīng)在url指定的對應(yīng)位置上了。在job在node上執(zhí)行之前，DistributedCache會(huì)copy必要的文件到這個(gè)slave node。它的功效就是為每個(gè)job只copy一次，而且copy到指定位置，能夠自動(dòng)解壓縮。

DistributedCache可以用來分發(fā)簡單的只讀文件，或者一些復(fù)雜的例如archive，jar文件等。archive文件會(huì)自動(dòng)解壓縮，而jar文件會(huì)被自動(dòng)放置到任務(wù)的classpath中（lib）。分發(fā)壓縮archive時(shí)，可以指定解壓名稱如：dict.zip#dict。這樣就會(huì)解壓到dict中，否則默認(rèn)是dict.zip中。

文件是有執(zhí)行權(quán)限的。用戶可以選擇在任務(wù)的工作目錄下建立指向DistributedCache的軟鏈接。

1.DistributedCache.createSymlink(conf);    

2.    DistributedCache.addCacheFile(new Path("hdfs://host:port/absolute-path#link-name").toUri(), conf);      

DistributedCache.createSymlink(Configuration)方法讓DistributedCache在當(dāng)前工作目錄下創(chuàng)建到緩存文件的符號(hào)鏈接。則在task的當(dāng)前工作目錄會(huì)有l(wèi)ink-name的鏈接，相當(dāng)于快捷方法，鏈接到expr.txt文件,在setup方法使用的情況則要簡單許多?；蛘咄ㄟ^設(shè)置配置文件屬性mapred.create.symlink為yes。分布式緩存會(huì)截取URI的片段作為鏈接的名字。例如，URI是hdfs://namenode:port/lib.so.1#lib.so，則在task當(dāng)前工作目錄會(huì)有名為lib.so的鏈接，它會(huì)鏈接分布式緩存中的lib.so.1

DistributedCache會(huì)跟蹤修改緩存文件的timestamp。

下面是使用的例子,為應(yīng)用app設(shè)置緩存    

1. 將需要的文件copy到FileSystem中：  

1.  $ bin/hadoop fs -copyFromLocal lookup.dat /myapp/lookup.dat    

2.  $ bin/hadoop fs -copyFromLocal map.zip /myapp/map.zip    

3.  $ bin/hadoop fs -copyFromLocal mylib.jar /myapp/mylib.jar  

4.  $ bin/hadoop fs -copyFromLocal mytar.tar /myapp/mytar.tar  

5.  $ bin/hadoop fs -copyFromLocal mytgz.tgz /myapp/mytgz.tgz  

6.  $ bin/hadoop fs -copyFromLocal mytargz.tar.gz /myapp/mytargz.tar.gz  

2. 設(shè)置app的jobConf：  

7.  JobConf job = new JobConf();  

8.  DistributedCache.addCacheFile(new URI("/myapp/lookup.dat#lookup.dat"),   

9.                                job);  

10.  DistributedCache.addCacheArchive(new URI("/myapp/map.zip", job);  

11.  DistributedCache.addFileToClassPath(new Path("/myapp/mylib.jar"), job);  

12.  DistributedCache.addCacheArchive(new URI("/myapp/mytar.tar", job);  

13.  DistributedCache.addCacheArchive(new URI("/myapp/mytgz.tgz", job);  

14.  DistributedCache.addCacheArchive(new URI("/myapp/mytargz.tar.gz", job);  

3. 在mapper或者reducer中使用緩存文件：  

15.  public static class MapClass extends MapReduceBase    

16.  implements Mapper<K, V, K, V> {  

17.    

18.    private Path[] localArchives;  

19.    private Path[] localFiles;  

20.      

21.    public void configure(JobConf job) {  

22.      // 得到剛剛緩存的文件  

23.      localArchives = DistributedCache.getLocalCacheArchives(job);  

24.      localFiles = DistributedCache.getLocalCacheFiles(job);  

25.    }  

26.      

27.    public void map(K key, V value,   

28.                    OutputCollector<K, V>; output, Reporter reporter)   

29.    throws IOException {  

30.      // 使用緩存文件  

31.      // ...  

32.      // ...  

33.      output.collect(k, v);  

34.    }  

35.  }  

 
 它跟GenericOptionsParser的部分功能有異曲同工之妙。

PathFilter +通配符。accept(Path path)篩選path是否通過。

名稱欄目：hadoop中mapreduce的常用類（一）
URL網(wǎng)址：http://aaarwkj.com/article18/jeshdp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供商城網(wǎng)站、移動(dòng)網(wǎng)站建設(shè)、網(wǎng)頁設(shè)計(jì)公司、做網(wǎng)站、微信小程序、App開發(fā)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)