1、MapReduce理論簡介
1.1 MapReduce編程模型
MapReduce采用"分而治之"的思想，把對(duì)大規(guī)模數(shù)據(jù)集的操作，分發(fā)給一個(gè)主節(jié)點(diǎn)管理下的各個(gè)分節(jié)點(diǎn)共同完成，然后通過整合各個(gè)節(jié)點(diǎn)的中間結(jié)果，得到最終結(jié)果。簡單地說，MapReduce就是"任務(wù)的分解與結(jié)果的匯總"。
在Hadoop中，用于執(zhí)行MapReduce任務(wù)的機(jī)器角色有兩個(gè)：一個(gè)是JobTracker；另一個(gè)是TaskTracker，JobTracker是用于調(diào)度工作的，TaskTracker是用于執(zhí)行工作的。一個(gè)Hadoop集群中只有一臺(tái)JobTracker。
在分布式計(jì)算中，MapReduce框架負(fù)責(zé)處理了并行編程中分布式存儲(chǔ)、工作調(diào)度、負(fù)載均衡、容錯(cuò)均衡、容錯(cuò)處理以及網(wǎng)絡(luò)通信等復(fù)雜問題，把處理過程高度抽象為兩個(gè)函數(shù)：map和reduce，map負(fù)責(zé)把任務(wù)分解成多個(gè)任務(wù)，reduce負(fù)責(zé)把分解后多任務(wù)處理的結(jié)果匯總起來。
需要注意的是，用MapReduce來處理的數(shù)據(jù)集（或任務(wù)）必須具備這樣的特點(diǎn)：待處理的數(shù)據(jù)集可以分解成許多小的數(shù)據(jù)集，而且每一個(gè)小數(shù)據(jù)集都可以完全并行地進(jìn)行處理。
1.2 MapReduce處理過程
在Hadoop中，每個(gè)MapReduce任務(wù)都被初始化為一個(gè)Job，每個(gè)Job又可以分為兩種階段：map階段和reduce階段。這兩個(gè)階段分別用兩個(gè)函數(shù)表示，即map函數(shù)和reduce函數(shù)。map函數(shù)接收一個(gè)<key,value>形式的輸入，然后同樣產(chǎn)生一個(gè)<key,value>形式的中間輸出，Hadoop函數(shù)接收一個(gè)如<key,(list of values)>形式的輸入，然后對(duì)這個(gè)value集合進(jìn)行處理，每個(gè)reduce產(chǎn)生0或1個(gè)輸出，reduce的輸出也是<key,value>形式的。
MapReduce處理大數(shù)據(jù)集的過程
2、運(yùn)行WordCount程序
單詞計(jì)數(shù)是最簡單也是最能體現(xiàn)MapReduce思想的程序之一，可以稱為MapReduce版"Hello World"，該程序的完整代碼可以在Hadoop安裝包的"src/examples"目錄下找到。單詞計(jì)數(shù)主要完成功能是：統(tǒng)計(jì)一系列文本文件中每個(gè)單詞出現(xiàn)的次數(shù)，如下圖所示。
2.1 準(zhǔn)備工作
現(xiàn)在以"hadoop"普通用戶登錄"Master.Hadoop"服務(wù)器。
1）創(chuàng)建本地示例文件
首先在"/home/hadoop"目錄下創(chuàng)建文件夾"file"。
接著創(chuàng)建兩個(gè)文本文件file1.txt和file2.txt，使file1.txt內(nèi)容為"Hello World"，而file2.txt的內(nèi)容為"Hello Hadoop"。
2）在HDFS上創(chuàng)建輸入文件夾
3）上傳本地file中文件到集群的input目錄下
2.2 運(yùn)行例子
1）在集群上運(yùn)行WordCount程序
備注：以input作為輸入目錄，output目錄作為輸出目錄。
已經(jīng)編譯好的WordCount的Jar在"/usr/hadoop"下面，就是"hadoop-examples-1.0.0.jar"，所以在下面執(zhí)行命令時(shí)記得把路徑寫全了，不然會(huì)提示找不到該Jar包。
2）MapReduce執(zhí)行過程顯示信息
Hadoop命令會(huì)啟動(dòng)一個(gè)JVM來運(yùn)行這個(gè)MapReduce程序，并自動(dòng)獲得Hadoop的配置，同時(shí)把類的路徑（及其依賴關(guān)系）加入到Hadoop的庫中。以上就是Hadoop Job的運(yùn)行記錄，從這里可以看到，這個(gè)Job被賦予了一個(gè)ID號(hào)：job_201202292213_0002，而且得知輸入文件有兩個(gè)（Total input paths to process : 2），同時(shí)還可以了解map的輸入輸出記錄（record數(shù)及字節(jié)數(shù)），以及reduce輸入輸出記錄。比如說，在本例中，map的task數(shù)量是2個(gè)，reduce的task數(shù)量是一個(gè)。map的輸入record數(shù)是2個(gè)，輸出record數(shù)是4個(gè)等信息。
2.3 查看結(jié)果
1）查看HDFS上output目錄內(nèi)容
從上圖中知道生成了三個(gè)文件，我們的結(jié)果在"part-r-00000"中。
2）查看結(jié)果輸出文件內(nèi)容
3、WordCount源碼分析
3.1 特別數(shù)據(jù)類型介紹
Hadoop提供了如下內(nèi)容的數(shù)據(jù)類型，這些數(shù)據(jù)類型都實(shí)現(xiàn)了WritableComparable接口，以便用這些類型定義的數(shù)據(jù)可以被序列化進(jìn)行網(wǎng)絡(luò)傳輸和文件存儲(chǔ)，以及進(jìn)行大小比較。
BooleanWritable：標(biāo)準(zhǔn)布爾型數(shù)值
ByteWritable：單字節(jié)數(shù)值
DoubleWritable：雙字節(jié)數(shù)
FloatWritable：浮點(diǎn)數(shù)
IntWritable：整型數(shù)
LongWritable：長整型數(shù)
Text：使用UTF8格式存儲(chǔ)的文本
NullWritable：當(dāng)<key,value>中的key或value為空時(shí)使用
3.2 舊的WordCount分析
1）源代碼程序
package org.apache.hadoop.examples;
import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
public class WordCount {
public static class Map extends MapReduceBase implements
Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(LongWritable key, Text value,
OutputCollector<Text, IntWritable> output, Reporter reporter)
throws IOException {
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
while (tokenizer.hasMoreTokens()) {
word.set(tokenizer.nextToken());
output.collect(word, one);
}
}
}
public static class Reduce extends MapReduceBase implements
Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values,
OutputCollector<Text, IntWritable> output, Reporter reporter)
throws IOException {
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
}
}
public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(WordCount.class);
conf.setJobName("wordcount");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
JobClient.runJob(conf);
}
}
3）主方法Main分析
public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(WordCount.class);
conf.setJobName("wordcount");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
JobClient.runJob(conf);
}
首先講解一下Job的初始化過程。main函數(shù)調(diào)用Jobconf類來對(duì)MapReduce
Job進(jìn)行初始化，然后調(diào)用setJobName()方法命名這個(gè)Job。對(duì)Job進(jìn)行合理的命名有助于更快地找到Job，以便在JobTracker和Tasktracker的頁面中對(duì)其進(jìn)行監(jiān)視。
JobConf conf = new JobConf(WordCount. class ); conf.setJobName("wordcount" );
接著設(shè)置Job輸出結(jié)果<key,value>的中key和value數(shù)據(jù)類型，因?yàn)榻Y(jié)果是<單詞,個(gè)數(shù)>，所以key設(shè)置為"Text"類型，相當(dāng)于Java中String類型。Value設(shè)置為"IntWritable"，相當(dāng)于Java中的int類型。
conf.setOutputKeyClass(Text.class );
conf.setOutputValueClass(IntWritable.class );
然后設(shè)置Job處理的Map（拆分）、Combiner（中間結(jié)果合并）以及Reduce（合并）的相關(guān)處理類。這里用Reduce類來進(jìn)行Map產(chǎn)生的中間結(jié)果合并，避免給網(wǎng)絡(luò)數(shù)據(jù)傳輸產(chǎn)生壓力。
conf.setMapperClass(Map.class );
conf.setCombinerClass(Reduce.class );
conf.setReducerClass(Reduce.class );
接著就是調(diào)用setInputPath()和setOutputPath()設(shè)置輸入輸出路徑。
conf.setInputFormat(TextInputFormat.class );
conf.setOutputFormat(TextOutputFormat.class );
（1）InputFormat和InputSplit
InputSplit是Hadoop定義的用來傳送給每個(gè)單獨(dú)的map的數(shù)據(jù)，InputSplit存儲(chǔ)的并非數(shù)據(jù)本身，而是一個(gè)分片長度和一個(gè)記錄數(shù)據(jù)位置的數(shù)組。生成InputSplit的方法可以通過InputFormat()來設(shè)置。
當(dāng)數(shù)據(jù)傳送給map時(shí)，map會(huì)將輸入分片傳送到InputFormat，InputFormat則調(diào)用方法getRecordReader()生成RecordReader，RecordReader再通過creatKey()、creatValue()方法創(chuàng)建可供map處理的<key,value>對(duì)。簡而言之，InputFormat()方法是用來生成可供map處理的<key,value>對(duì)的。
Hadoop預(yù)定義了多種方法將不同類型的輸入數(shù)據(jù)轉(zhuǎn)化為map能夠處理的<key,value>對(duì)，它們都繼承自InputFormat，分別是：
InputFormat
|
|---BaileyBorweinPlouffe.BbpInputFormat
|---ComposableInputFormat
|---CompositeInputFormat
|---DBInputFormat
|---DistSum.Machine.AbstractInputFormat
|---FileInputFormat
|---CombineFileInputFormat
|---KeyValueTextInputFormat
|---NLineInputFormat
|---SequenceFileInputFormat
|---TeraInputFormat
|---TextInputFormat
其中TextInputFormat是Hadoop默認(rèn)的輸入方法，在TextInputFormat中，每個(gè)文件（或其一部分）都會(huì)單獨(dú)地作為map的輸入，而這個(gè)是繼承自FileInputFormat的。之后，每行數(shù)據(jù)都會(huì)生成一條記錄，每條記錄則表示成<key,value>形式：
value值是每行的內(nèi)容，數(shù)據(jù)類型是Text。
?。?）OutputFormat
每一種輸入格式都有一種輸出格式與其對(duì)應(yīng)。默認(rèn)的輸出格式是TextOutputFormat，這種輸出方式與輸入類似，會(huì)將每條記錄以一行的形式存入文本文件。不過，它的鍵和值可以是任意形式的，因?yàn)槌绦?span style="margin:0px; padding:0px; line-height:1.8; color:red">內(nèi)容會(huì)調(diào)用toString()方法將鍵和值轉(zhuǎn)換為String類型再輸出。
3）Map類中map方法分析
public static class Map extends MapReduceBase implements
Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(LongWritable key, Text value,
OutputCollector<Text, IntWritable> output, Reporter reporter)
throws IOException {
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
while (tokenizer.hasMoreTokens()) {
word.set(tokenizer.nextToken());
output.collect(word, one);
}
}
}
Map類繼承自MapReduceBase，并且它實(shí)現(xiàn)了Mapper接口，此接口是一個(gè)規(guī)范類型，它有4種形式的參數(shù)，分別用來指定map的輸入key值類型、輸入value值類型、輸出key值類型和輸出value值類型。在本例中，因?yàn)槭褂玫氖荰extInputFormat，它的輸出key值是LongWritable類型，輸出value值是Text類型，所以map的輸入類型為<LongWritable,Text>。在本例中需要輸出<word,1>這樣的形式，因此輸出的key值類型是Text，輸出的value值類型是IntWritable。
實(shí)現(xiàn)此接口類還需要實(shí)現(xiàn)map方法，map方法會(huì)具體負(fù)責(zé)對(duì)輸入進(jìn)行操作，在本例中，map方法對(duì)輸入的行以空格為單位進(jìn)行切分，然后使用OutputCollect收集輸出的<word,1>。
4）Reduce類中reduce方法分析
public static class Reduce extends MapReduceBase implements
Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values,
OutputCollector<Text, IntWritable> output, Reporter reporter)
throws IOException {
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
}
}
Reduce類也是繼承自MapReduceBase的，需要實(shí)現(xiàn)Reducer接口。Reduce類以map的輸出作為輸入，因此Reduce的輸入類型是<Text，Intwritable>。而Reduce的輸出是單詞和它的數(shù)目，因此，它的輸出類型是<Text,IntWritable>。Reduce類也要實(shí)現(xiàn)reduce方法，在此方法中，reduce函數(shù)將輸入的key值作為輸出的key值，然后將獲得多個(gè)value值加起來，作為輸出的值。
3.3 新的WordCount分析
1）源代碼程序
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();
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);
}
}
Map過程需要繼承org.apache.hadoop.mapreduce包中Mapper類，并重寫其map方法。通過在map方法中添加兩句把key值和value值輸出到控制臺(tái)的代碼，可以發(fā)現(xiàn)map方法中value值存儲(chǔ)的是文本文件中的一行（以回車符為行結(jié)束標(biāo)記），而key值為該行的首字母相對(duì)于文本文件的首地址的偏移量。然后StringTokenizer類將每一行拆分成為一個(gè)個(gè)的單詞，并將<word,1>作為map方法的結(jié)果輸出，其余的工作都交有MapReduce框架處理。
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);
}
}
Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reducer類，并重寫其reduce方法。Map過程輸出<key,values>中key為單個(gè)單詞，而values是對(duì)應(yīng)單詞的計(jì)數(shù)值所組成的列表，Map的輸出就是Reduce的輸入，所以reduce方法只要遍歷values并求和，即可得到某個(gè)單詞的總次數(shù)。
3）執(zhí)行MapReduce任務(wù)
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
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);
}
在MapReduce中，由Job對(duì)象負(fù)責(zé)管理和運(yùn)行一個(gè)計(jì)算任務(wù)，并通過Job的一些方法對(duì)任務(wù)的參數(shù)進(jìn)行相關(guān)的設(shè)置。此處設(shè)置了使用TokenizerMapper完成Map過程中的處理和使用IntSumReducer完成Combine和Reduce過程中的處理。還設(shè)置了Map過程和Reduce過程的輸出類型：key的類型為Text，value的類型為IntWritable。任務(wù)的輸出和輸入路徑則由命令行參數(shù)指定，并由FileInputFormat和FileOutputFormat分別設(shè)定。完成相應(yīng)任務(wù)的參數(shù)設(shè)定后，即可調(diào)用job.waitForCompletion()方法執(zhí)行任務(wù)。
4、WordCount處理過程
本節(jié)將對(duì)WordCount進(jìn)行更詳細(xì)的講解。詳細(xì)執(zhí)行步驟如下：
1）將文件拆分成splits，由于測試用的文件較小，所以每個(gè)文件為一個(gè)split，并將文件按行分割形成<key,value>對(duì)，如圖4-1所示。這一步由MapReduce框架自動(dòng)完成，其中偏移量（即key值）包括了回車所占的字符數(shù)（Windows和Linux環(huán)境會(huì)不同）。
圖4-1 分割過程
2）將分割好的<key,value>對(duì)交給用戶定義的map方法進(jìn)行處理，生成新的<key,value>對(duì)，如圖4-2所示。
圖4-2 執(zhí)行map方法
3）得到map方法輸出的<key,value>對(duì)后，Mapper會(huì)將它們按照key值進(jìn)行排序，并執(zhí)行Combine過程，將key至相同value值累加，得到Mapper的最終輸出結(jié)果。如圖4-3所示。
圖4-3 Map端排序及Combine過程
4）Reducer先對(duì)從Mapper接收的數(shù)據(jù)進(jìn)行排序，再交由用戶自定義的reduce方法進(jìn)行處理，得到新的<key,value>對(duì)，并作為WordCount的輸出結(jié)果，如圖4-4所示。
圖4-4 Reduce端排序及輸出結(jié)果
5、MapReduce新舊改變
Hadoop最新版本的MapReduce Release 0.20.0的API包括了一個(gè)全新的Mapreduce JAVA API，有時(shí)候也稱為上下文對(duì)象。
新的API類型上不兼容以前的API，所以，以前的應(yīng)用程序需要重寫才能使新的API發(fā)揮其作用 。
新的API和舊的API之間有下面幾個(gè)明顯的區(qū)別。
-
新的API傾向于使用抽象類，而不是接口，因?yàn)檫@更容易擴(kuò)展。例如，你可以添加一個(gè)方法(用默認(rèn)的實(shí)現(xiàn))到一個(gè)抽象類而不需修改類之前的實(shí)現(xiàn)方法。在新的API中，Mapper和Reducer是抽象類。
-
新的API是在org.apache.hadoop.mapreduce包(和子包)中的。之前版本的API則是放在org.apache.hadoop.mapred中的。
-
新的API廣泛使用context object(上下文對(duì)象)，并允許用戶代碼與MapReduce系統(tǒng)進(jìn)行通信。例如，MapContext基本上充當(dāng)著JobConf的OutputCollector和Reporter的角色。
-
新的API同時(shí)支持"推"和"拉"式的迭代。在這兩個(gè)新老API中，鍵/值記錄對(duì)被推mapper中，但除此之外，新的API允許把記錄從map()方法中拉出，這也適用于reducer。"拉"式的一個(gè)有用的例子是分批處理記錄，而不是一個(gè)接一個(gè)。
-
新的API統(tǒng)一了配置。舊的API有一個(gè)特殊的JobConf對(duì)象用于作業(yè)配置，這是一個(gè)對(duì)于Hadoop通常的Configuration對(duì)象的擴(kuò)展。在新的API中，這種區(qū)別沒有了，所以作業(yè)配置通過Configuration來完成。作業(yè)控制的執(zhí)行由Job類來負(fù)責(zé)，而不是JobClient，它在新的API中已經(jīng)蕩然無存。
|
