标签:
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); } } }
标签:
原文地址:http://www.cnblogs.com/ciade/p/5009638.html