mapreduce是hadoop的核心部分之一全度。是分布式運(yùn)算程序的編程框架肋演。相對于hdfs阅仔，mapreduce就是一個(gè)客戶端。
hdfs：
namenode,管理整個(gè)系統(tǒng)的元數(shù)據(jù)
datanode,管理用戶的文件數(shù)據(jù)塊（不負(fù)責(zé)切片台谊，切片客戶端完成）蓉媳，
每一個(gè)文件塊有多個(gè)副本，存放于不同的datanode上锅铅。
定期向namenode匯報(bào)自身保存的文件block信息酪呻，namenode會(huì)負(fù)責(zé)保持文件副本的數(shù)量。
secondarydatanode狠角，做checkpoint号杠。
hdfs不支持文件更改內(nèi)容，只能追加。
分布式運(yùn)算程序一般分為兩個(gè)階段姨蟋。第一階段map的并發(fā)實(shí)例maptask完全并行屉凯。第二階段reduce的reducetask也是互不相干。但是他們的數(shù)據(jù)依賴于上一階段maptask的輸出眼溶。mapreduce編程模型只能包含一個(gè)map和一個(gè)reduce悠砚。如果業(yè)務(wù)邏輯復(fù)雜，只能多個(gè)mapreduce程序串行運(yùn)行堂飞。
maptask和reducetask由mr application master協(xié)調(diào)灌旧。
- Wordcount例子
package hadoop.wcdemo;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
String line = value.toString();
String[] words = line.split(" ");
for(String word:words){
context.write(new Text(word), new IntWritable(1));
}
}
}
package hadoop.wcdemo;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
@Override
protected void reduce(Text key, Iterable<IntWritable> values,Context context)
throws IOException, InterruptedException {
int count=0;
for(IntWritable value:values){
count+=value.get();
}
context.write(key, new IntWritable(count));
}
}
package hadoop.wcdemo;
import java.io.IOException;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordcountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
/*conf.set("mapreduce.framework.name", "yarn");
conf.set("yarn.resourcemanager.hostname", "mini1");*/
Job job = Job.getInstance(conf);
job.setJarByClass(WordcountDriver.class);
job.setMapperClass(WordcountMapper.class);
job.setReducerClass(WordcountReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
/*job.submit();*/
boolean res = job.waitForCompletion(true);
System.exit(res?0:1);
}
}
在集群上運(yùn)行：
先把編好的程序打成jar包上傳到集群任一機(jī)器
start-dfs.sh
start-yarn.sh
hadoop fs -mkdir -p /wordcount/input
hadoop fs -put xxx /wordcount/input
hadoop jar wordcount.jar hadoop.wcdemo.WordcountDriver /wordcount/input /wordcount/output
程序分為3個(gè)部分，Mapper铝噩，Reducer（繼承父類），Driver（提交運(yùn)行mr程序的客戶端）毛甲。Mapper輸入kv對具被，輸出也是kv對的形式。Mapper的業(yè)務(wù)邏輯寫在map方法中七咧。map方法（maptask進(jìn)程）對每一個(gè)<k,v>調(diào)用一次（k是那一行起始偏移量v是一行的內(nèi)容坑雅，對應(yīng)一行）衬横。Reducer輸入類型對應(yīng)Mapper的輸出數(shù)據(jù)類型蜂林。業(yè)務(wù)邏輯寫在reduce方法中拇泣。reducetask進(jìn)程對每一組相同k的<k,v>組調(diào)用一次reduce方法。Driver提交的是一個(gè)描述了各種必要信息的job對象睁蕾。
submit后先查看hdfs獲取待處理文件的信息子眶，根據(jù)參數(shù)配置形成任務(wù)分配的規(guī)劃（文件的分片）（job.split,job.xml,wc.jar），然后把這些對象提交給yarn粤咪，找一臺機(jī)器啟動(dòng)mr appmaster渴杆。mr appmaster根據(jù)文件的分片啟動(dòng)maptask進(jìn)程（優(yōu)先在存文件的機(jī)器上起相應(yīng)的maptask）。maptask進(jìn)程其實(shí)是一個(gè)管理者磁奖，調(diào)用很多組件來完成任務(wù)囊拜。一行一行讀文件是其實(shí)是調(diào)用inputformat組件里的方法。讀到的kv結(jié)果后再調(diào)用我們定義的wordcountmapper（map(k,v),context.write(k,v)）比搭。再交給outputcollector組件收集文件排序分區(qū)（按照給那個(gè)reducer分區(qū)）冠跷。所有maptask執(zhí)行完后，mr appmaster再啟動(dòng)reducetask敢辩。reducetask從剛才maptask寫好的分區(qū)文件中取出屬于自己分區(qū)的數(shù)據(jù)蔽莱。每一組kv對調(diào)用wordcountreducer（reduce(k,itvalues),context.write(k,v)）。最后調(diào)用outputformat組件寫出數(shù)據(jù)（不需要收集緩存直接寫出）戚长。往hdfs文件（part-r00001,part-r00002……）不斷追加盗冷。
