如何構(gòu)建MapReduce程序的基礎(chǔ)模板

本篇內(nèi)容主要講解“如何構(gòu)建MapReduce程序的基礎(chǔ)模板”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“如何構(gòu)建MapReduce程序的基礎(chǔ)模板”吧!

什么是開發(fā)數(shù)據(jù)集？

一個(gè)流行的開發(fā)策略是為生產(chǎn)環(huán)境中的大數(shù)據(jù)集建立一個(gè)較小的、抽樣的數(shù)據(jù)子集，稱為開發(fā)數(shù)據(jù)集。這個(gè)開發(fā)數(shù)據(jù)集可能只有幾百兆字節(jié)。當(dāng)你以單機(jī)或者偽分布式模式編寫程序來(lái)處理它們時(shí)，你會(huì)發(fā)現(xiàn)開發(fā)周期很短，在自己的機(jī)器上運(yùn)行程序也很方便，而且還可以在獨(dú)立的環(huán)境中進(jìn)行調(diào)試。

為什么選擇專利引用數(shù)據(jù)做測(cè)試？

1、因?yàn)樗鼈兣c你將來(lái)會(huì)遇到的大多數(shù)數(shù)據(jù)類型相似

2、專利引用數(shù)據(jù)所構(gòu)成的關(guān)系圖與網(wǎng)頁(yè)鏈接以及社會(huì)網(wǎng)絡(luò)圖可謂大同小異

3、專利發(fā)布以時(shí)間為序，有些特性類似于時(shí)間序列

4、每個(gè)專利關(guān)聯(lián)到一個(gè)人 (發(fā)明人) 和一個(gè)位置 (發(fā)明人的國(guó)家)，你可以將之視為個(gè)人信息或地理數(shù)據(jù)

5、你可以將這些數(shù)據(jù)視為具有明確模式的普通數(shù)據(jù)庫(kù)關(guān)系，而格式上簡(jiǎn)單地以逗號(hào)分開

數(shù)據(jù)集采用標(biāo)準(zhǔn)

數(shù)據(jù)集采用標(biāo)準(zhǔn)的逗號(hào)分隔取值 (comma-separated values, CSV) 格式。

構(gòu)建MapReduce程序的基礎(chǔ)模板

大多數(shù)MapReduce程序的編寫都可以簡(jiǎn)單地依賴于一個(gè)模板及其變種，當(dāng)撰寫一個(gè)新得MapReduce程序時(shí)，我們通常會(huì)采用一個(gè)現(xiàn)有的MapReduce程序，并將其修改成我們所希望的樣子。

典型的Hadoop程序的模板

public class MyJob extends Configured implements Tool {

       public static class MapClass extends MapReduceBase

              implements Mapper<Text, Text, Text, Text> {

              public void map (Text key, Text value,

                                   OutputCollector<Text, Text> output,

                                   Reporter reporter) throws IOException {

                     output.collect(value, key);

              }

       }

       public static class Reduce extends MapReduceBase

              implements Reducer<Text, Text, Text, Text> {

              public void reduce(Text key, Iterator<Text> values,

                                   OutputCollector<Text, Text> output,

                                   Reporter reporter) throws IOException {

                     String csv = "";

                     while (values.hasNext()) {

                            if (csv.length() > 0) csv += ",";

                            csv += values.next().toString();

                     }

                     output.collect(key, new Text(csv));

              }

       }

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

              Configuration conf = getConf();

              JobConf job = new JobConf(conf, MyJob.class);

              Path in = new Path(args[0]);

              Path out = new Path(args[1]);

              FileInputFormat.setInputPaths(job, in);

              FileOutputFormat.setOutputPath(job, out);

              job.setJobName("MyJob");

              job.setMapperClass(MapClass.class);

              job.setReducerClass(Reduce.class);

              job.setInputFormat(KeyValueTextInputFormat.class);

              job.setOutputFormat(TextOutputFormat.class);

              job.setOutputKeyClass(Text.class);

              job.setOutputValueClass(Text.class);

              job.set("key.value.separator.in.input.line", ",");

              JobClient.runJob(job);

              return 0;

       }

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

              int res = ToolRunner.run(new Configuration(), new MyJob(), args);

              System.exit(res);

       }

}

1、我們習(xí)慣用單個(gè)類來(lái)完整地定義每個(gè)MapReduce作業(yè)，這里成為MyJob類

2、Hadoop要求Mapper和Reducer必須是它們自身的靜態(tài)類，這些類非常小，模板將它們包含在MyJob類中作為內(nèi)部類，這樣做的好處是可以把所有的東西放在一個(gè)文件內(nèi)，簡(jiǎn)化代碼管理

3、但是需要記住這些內(nèi)部類是獨(dú)立的，通常不與MyJob類進(jìn)行交互

4、在作業(yè)執(zhí)行期間，采用不同JVM的各類節(jié)點(diǎn)復(fù)制并運(yùn)行Mapper和Reducer，而其他的作業(yè)類僅在客戶機(jī)上執(zhí)行

解釋下run()方法

1、框架的核心在run()方法中，也稱為driver

2、它實(shí)例化、配置并傳遞一個(gè)JobConf對(duì)象命名的作業(yè)給JobClient.runJob()以啟動(dòng)MapReduce作業(yè)(反過(guò)來(lái)，JobClient類與JobTracker通信讓該作業(yè)在集群上啟動(dòng))

3、JobConf對(duì)象將保持作業(yè)運(yùn)行所需的全部配置參數(shù)

4、Driver需要在作業(yè)中為每個(gè)作業(yè)定制基本參數(shù)，包括輸入路徑、輸出路徑、Mapper類和Reducer類

5、每個(gè)作業(yè)可以重置默認(rèn)的作業(yè)屬性，例如，InputFormat、OutputFormat等，也可以調(diào)用JobConf對(duì)象中的set()方法填充任意的配置參數(shù)

6、一旦傳遞JobConf對(duì)象到JobClient.runJob()，他就被視為決定這個(gè)作業(yè)如何運(yùn)行的藍(lán)本

關(guān)于driver的配置的一些說(shuō)明

1、JobConf對(duì)象有許多參數(shù)，但我們并不希望全部的參數(shù)都通過(guò)編寫driver來(lái)設(shè)置，可以把Hadoop安裝時(shí)的配置文件作為一個(gè)很好的起點(diǎn)

2、用戶可能希望在命令行啟動(dòng)一個(gè)作業(yè)時(shí)傳遞額外的參數(shù)來(lái)改變作業(yè)配置

3、Driver可以通過(guò)自定義一組命令并自行處理用戶參數(shù)，來(lái)支持用戶修改其中的一些配置

4、因?yàn)榻?jīng)常需要做這樣的任務(wù)，Hadoop框架便提供了ToolRunner、Tool和Configured來(lái)簡(jiǎn)化其實(shí)現(xiàn)。

5、當(dāng)它們?cè)谏厦娴腗yJob框架中被同時(shí)使用時(shí)，這些類使得作業(yè)可以理解用戶提供的被GenericOptionParser支持的選項(xiàng)

比如下面的命令：

bin/hadoop jar playgroup/MyJob.jar MyJob input/cite75-99.txt output

如果我們運(yùn)行作業(yè)僅僅是想看到mapper的輸出 (處于調(diào)試的目的), 可以用選項(xiàng) -D mapred.reduce.tasks=0將reducer的數(shù)目設(shè)置為0

bin/hadoop jar playgroup/MyJob.jar MyJob -D mapred.reduce.tasks=0 input/cite75-99.txt output

通過(guò)使用ToolRunner、MyJob可以自動(dòng)支持一下選項(xiàng)
GenericOptionsParser支持的選項(xiàng)

模板代碼Mappper與Reducer

模板中習(xí)慣將Mapper類稱為MapClass，而將Reducer類稱為Reduce

Mapper和Reducer都是MapReduceBase的擴(kuò)展

MapReduceBase是個(gè)小類，包含configure()和close()，我們使用上兩個(gè)方法來(lái)建立和清除map(reduce)任務(wù)，除非是更高級(jí)的作業(yè)，通常我們并不需要覆蓋它們

Mapper類和Reducer類模板說(shuō)明

public static class MapClass extends MapReduceBase

       implements Mapper<K1, V1, K2, V2> {

       public void map (K1 key, V1 value,

                                   OutputCollector<K2, V2> output,

                                   Reporter reporter) throws IOException { }

}

public static class Reduce extends MapReduceBase

       implements Reducer<K1, V2, K3, V3> {

       public void reduce(K2 key, Iterator<V2> values,

                                   OutputCollector<K3, V3> output,

                                   Reporter reporter) throws IOException { }

}

Mapper類的核心操作為map()方法，Reduce類為reduce()方法。每一個(gè)map()方法的調(diào)用分別被賦予一個(gè)類型為K1和V1的鍵/值對(duì)。這個(gè)鍵/值對(duì)由mapper生成，并通過(guò)OutputCollector對(duì)象的collect()方法來(lái)輸出。你需要在map()方法中的合適位置調(diào)用:

output.collect((K2) k, (V2) v);

在Reducer中reduce()方法的每次調(diào)用均被賦予K2類型的鍵，以及V2類型的一組值。注意它必須與Mapper中使用的K2和V2類型相同。Reduce()方法可能會(huì)循環(huán)遍歷V2類型的所有值。

while (values.hasNext()) {

       V2 v = values.next();

}

Reduce()方法還使用OutputCollector來(lái)搜集其鍵/值的輸出，它們的類型為K3/V3。在reudce()方法中可以調(diào)用

output.collect((K3) k, (V3) v);

除了在Mapper和Reducer之間保持K2與V3的類型一致，還需要確保在Mapper和Reducer中使用的鍵值類型與在driver中設(shè)置的輸入格式、輸出鍵的類，以及輸出值的類保持一致

使用KeyValueTextInputFormat意味著K1和V1必須均為Text類型

Driver則必須調(diào)用setOutputKeyClass()和setOutputValueClass()分別指定K2和V2的類

最終：

1、所有的鍵與值的類型必須是Writable的子類型，來(lái)確保Hadoop的序列化接口可以把數(shù)據(jù)在分布式集群上發(fā)送

2、鍵的類型實(shí)現(xiàn)了WritableComparable，它是Writable的子接口，鍵的類型還需額外支持compareTo()方法，因?yàn)樵贛apReduce框架中鍵會(huì)被用來(lái)進(jìn)行排序

到此，相信大家對(duì)“如何構(gòu)建MapReduce程序的基礎(chǔ)模板”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！