MR操作————Map、Partitioner、Shuffle、Combiners、Reduce
1.Map步骤
1.1 读取输入文件，解析成k-v对，其中每个k-v对调用一次map函数
1.2 写自己的逻辑，对输入的k-v进行处理，转换成新的k-v
1.3 对输出的k-v进行分区(Partitioner)
1.4 对不同分区的数据进行排序/分组，将相同的key的value放在一个集合中(Shuffle处理)
1.5 分组后进行归约（可选）(Combiners 可理解为单个节点的reduce 不是所有算法都适合 如求平均数时就不适合，有可能造成结果不对)
2.Reduce步骤
2.1 对多个map输出数据按照不同的分区分配到不同的reduce节点
2.2 对多个map任务输出进行合并、排序。写自己的逻辑，对输入的k-v对进行处理，转换成新的k-v,每个k-v对调用一次reduce函数
2.3 把reduce结果保存到文件中输出
3.流程图：
文件输入
->Map提取相应数据(按照一定格式)
->Partitioner(按照key进行分区)
->Shuffle(按照分区进行排序)
->Combiners(按照分区进行归约)
->推送到各个Redeuce节点
->Reduce(根据输入数据按照逻辑进行运行)
文件输出
4.JavaAPI
步骤:
4.1 定义一个类继承Mapper，其中泛型四个参数 分别是输入key,输入value,输出key,输出value(注意类型需使用Hadoop中的数据类型，参见java类型对应Hadoop类型)
4.2 重写map()方法
4.3 定义一个类继承Reducer，其中泛型四个参数 分别是输入key,输入value,输出key,输出value(注意此时输入参数应该是Mapper的输出参数)
4.4 重写reduce()方法
4.5 编写启动任务
1.得到Configuration对象
2.得到Job对象
3.根据MR步骤设置Job参数
第一步：输入文件
输入目录
输入数据类型处理
第二步：自定义Mapper类
输入自定义Mapper类
输出数据类型(K-V)处理
第三步：(可不选)
分区
第四步：(可不选)
排序分组
第五步：(可不选)
归约
第六步：(可不选)
分配到不同Reduce节点
第七步:自定义Reduce类
输出自定义Reduce类
输出数据类型(K-V)处理
第八步：输出文件
输出路径
输出格式类
4.6 打包发布运行
5.Hadoop内置数据类型
5.1 基础数据类型
1.BooleanWritable:标准布尔型数值
2.ByteWritable:单字节数值
3.DoubleWritable:双字节数值
4.FloatWritable:浮点数
5.IntWritable:整型数
6.LongWritable:长整型数
7.Text:使用UTF8格式存储的文本
8.NullWritable:当<key, value>中的key或value为空时使用
5.2 自定义数据类型
1.继承接口Writable,实现其方法write()和readFields(), 以便该数据能被序列化后完成网络传输或文件输入/输出
2.如果该数据需要作为主键key使用，或需要比较数值大小时，则需要实现WritalbeComparable接口,实现其方法write(),readFields(),CompareTo()
6.MR中常用算法
1.单词计数
2.数据去重
3.排序
4.选择
5.分组
6.单表关联
7.多表连接
JavaAPI 示例代码：
package testHadoopMR;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class testHadoopMR {
static final String INPUT_PATH = "hdfs://hadoop:9000/input/file.ini"; //输入文件
static final String OUTPUT_PATH = "hdfs://hadoop:9000/out";//输出路径
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();//得到configuration对象
Job job = new Job(conf, "word count");//得到job对象
job.setJarByClass(testHadoopMR.class);//设置任务加载类
FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));//1.设置输入文件路径
job.setMapperClass(MyMapper.class);//2.自定义mapper
job.setMapOutputKeyClass(Text.class);//输出key类型
job.setMapOutputValueClass(IntWritable.class);//输出value类型
job.setPartitionerClass(HashPartitioner.class);//3.设置分区
job.setNumReduceTasks(1);//分区数为1
job.setGroupingComparatorClass(MyGroupingComparator.class);//4.设置排序和分组 需自定义排序分组类，不然会按照hash值排序
job.setCombinerClass(MyReducer.class);// 5.设置Map规约Combiner
job.setReducerClass(MyReduce.class);//7.自定义reduce
job.setOutputKeyClass(Text.class);//设置输出 key类型
job.setOutputValueClass(IntWritable.class);//设置输出value类型
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));//8.设置输出路径
job.waitForCompletion(true);
}
public static class MyMapper extends 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, 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 MyReduce 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);
}
}
}