MR相关内容

InputFormat（负责切片和为mapTask读取数据）
分区（负责将mapTask阶段的KV，按照分区进入不同的ReduceTask。分区数量小于等于ReduceTask数量，即ReduceTask可能处理到无数据的分区，但是分区不能向不存在的ReduceTask进行输出）
Key排序
合并Combiner（发生的时刻（两次）：快速排序后、第一次归并后、不能影响执行结果，输入kv与输出kv类型一致）
分组（第二次归并后，Reducer输入前，自定义分组）

切片（InputFormat）

类型	切片方法`getSplits`	kv方法`createRecordReader`	用途说明
`TextInputFormat`	FIF的切片方法	`LineRecordReader`	按照块大小分片，按行行读取记录。
`KeyValueTextInputFormat`	FIF的切片方法	`KeyValueLineRecordReader`	按照块大小分片，按行读取记录。
`SequenceFileInputFormat`	FIF的切片方法	`SequenceFileRecordReader`	按照块大小分片，专用读取上一个任务使用`SequenceFileOutputFormat`输出的文件。
`FixedLengthInputFormat`	FIF的切片方法	`FixedLengthRecordReader`	按照块大小分片，定长读取记录。(使用频率低)
`NLineInputFormat`	自定义，N行一片	`LineRecordReader`	通过指定行数进行分片，按行读取记录。
`CombineFileInputFormat`	自定义	`LineRecordReader`	合并小文件进行分片读取，按行读取记录。

RecordReader分类

类型	说明
`LineRecordReader`	按行读取。 key：`LongWritable`,内容偏移量 value：`Text`，一行数据
`KeyValueLineRecordReader`	按照指定分割符进行分割 key：`Text`,分割的前一部分 value：`Text`，分割的后一部分，可以通过configuration中的`mapreduce.input.keyvaluelinerecordreader.key.value.separator`属性指定，默认是`\t`
`FixedLengthRecordReader`	读取固定长度内容（byte） key：`LongWritable`，记录偏移量 value：`BytesWritable`，二进制数据
`SequenceFileRecordReader`	主要串联job执行,读取上一个job的结果，作为当前job的输入，可以传递对象数据

自定义InputFormat

默认的InputFormat是TextOutputFormat

场景：
假设目前有一堆小文件，需要通过一个MR程序转换为一个SequenceFile文件，其中key：文件路径，value：文件内容
思路：

自定义一个Format类继承FileInputFormat，其中key:Text，value:BytesWritable(因为文件内容不一定是文本)

需要重写方法：public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context)，此时我们需要定义一个自己的RecordReader

同时考虑分片问题，我们FileInputFormat默认是按照文件和块大小分片的，这里我们需要同一个文件不被切片即一个文件在一个分片内，需要重写方法protected boolean isSplitable(JobContext context, Path filename)

InputFormat.java

/*** 将一堆小文件，转换为一个SequenceFile文件，key：原文件路径，value：原文件内容* 这里的泛型即为MapTask的记录的输入类型，所以key:Text,value:BytesWritable(因为文件不一定是文本，所以用BytesWritable)*/
public class PackageInputFormat extends FileInputFormat<Text, BytesWritable> {@Overridepublic RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {return new PackageSequenceRecordReader();}@Overrideprotected boolean isSplitable(JobContext context, Path filename) {return false;}
}

RecordReader.java

public class PackageSequenceRecordReader extends RecordReader<Text, BytesWritable> {//标识文件是否已经被读取过private boolean notRead = true;private FileSplit fs = null;private Text key = new Text();private BytesWritable val = new BytesWritable();private FSDataInputStream inputStream;//初始化只执行一次@Overridepublic void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {fs = (FileSplit) split;Path path = fs.getPath();FileSystem fileSystem = path.getFileSystem(context.getConfiguration());inputStream =  fileSystem.open(path);}//是否还有下一个数据，返回true则标识还有数据，否则无数据@Overridepublic boolean nextKeyValue() throws IOException, InterruptedException {//判断是否已读取过，未读取过进行读取if(notRead){//设置keykey.set(fs.getPath().getName());//设置vallong fileLength = fs.getLength();System.out.println("fs.getlength():"+fileLength);byte[] buf = new byte[(int) fileLength];int length = inputStream.read(buf);val.set(buf,0,length);notRead = false;return true;}else{return false;}}//获取当前key@Overridepublic Text getCurrentKey() throws IOException, InterruptedException {return key;}//获取当前value@Overridepublic BytesWritable getCurrentValue() throws IOException, InterruptedException {return val;}//去读进度@Overridepublic float getProgress() throws IOException, InterruptedException {return notRead ? 0 : 1;}//关闭资源@Overridepublic void close() throws IOException {IOUtils.closeStream(inputStream);}
}

Driver.java

public class PackageDriver {public static void main(String[] args) throws Exception {//1. 创建一个JobConfiguration conf = new Configuration();conf.set("mapred.reduce.child.java.opts", "-Xmx512m");Job job = Job.getInstance(conf);//2. 设置类路径job.setJarByClass(FlowDriver.class);//3. 设置Mapper和Reducer：不需要MR//4. 设置Mapper和Reducer的输出类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(BytesWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(BytesWritable.class);//5. 设置输入输出数据FileInputFormat.setInputPaths(job,new Path("d://hadoop-study/inputformat/input"));FileOutputFormat.setOutputPath(job,new Path("d://hadoop-study/inputformat/output"));//设置InputFormatjob.setInputFormatClass(PackageInputFormat.class);//设置OutputFormatjob.setOutputFormatClass(SequenceFileOutputFormat.class);//6. 提交jobboolean b = job.waitForCompletion(true);System.exit(b?0:1);}
}

自定义分区

MR默认分区：HashPartitioner

public class HashPartitioner<K, V> extends Partitioner<K, V> {/** Use {@link Object#hashCode()} to partition. */public int getPartition(K key, V value,int numReduceTasks) {return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;}}

场景
将下列手机号按照号段分别打印在不同的文件中
内容
13012345678
13098765432
13912345678
13988888888
18701100168
16701100183
要求：
130一组
139一组
其他一组
解析：
通过自定义分区处理，将手机号码进行分区输出到不同的ReduceTask即可

重点：PhonePartition.java

public class PhonePartition extends Partitioner<Text, NullWritable> {@Overridepublic int getPartition(Text text, NullWritable nullWritable, int numPartitions) {String phone = text.toString();//截取前三位switch (phone.substring(0,3)){case "130" : return 0;case "139" : return 1;default:return 2;}}
}

Driver.java/Mapper.java/Reducer.java
重点

设置自定义分区Class
设置启动ReduceTask的数量，默认是1个，具体数量根据具体业务情况定义，可以大于分区数，但是不能小于，另外ReduceTask数量大于分区数量的话，由于存在部分ReduceTask接不到数据，会造成性能浪费。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;import java.io.IOException;public class PartitionDriver {public static void main(String[] args) throws Exception {// 创建一个jobConfiguration conf = new Configuration();Job job = Job.getInstance(conf);//设置job JarClasspathjob.setJarByClass(PartitionDriver.class);//设置Mapper 和Reduce classjob.setMapperClass(Mapper.class);job.setReducerClass(Reducer.class);//设置map输出和Reduce输出job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);/*-------------------重点 BEGIN --------------*///设置自定义分区job.setPartitionerClass(PhonePartition.class);//根据此处需要分三个文件，设置启动reduceTask数量job.setNumReduceTasks(3);/*-------------------重点  END  --------------*///设置输入输出路径FileInputFormat.setInputPaths(job,new Path("D:\\hadoop-study\\partition\\input"));FileOutputFormat.setOutputPath(job,new Path("D:\\hadoop-study\\partition\\output"));boolean b = job.waitForCompletion(true);System.exit( b ? 0 : 1);}private static class Mapper extends org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {context.write(value,NullWritable.get());}}private static class Reducer extends org.apache.hadoop.mapreduce.Reducer<Text,NullWritable,Text,NullWritable> {@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {context.write(key,NullWritable.get());}}
}

自定义Key排序

排序默认WritableComparator字典序排序

可以通过将Key实现WritableComparable接口来实现key排序

案例
输入姓名(空格)成绩
张三 60
李四 80
王五 70
赵六 90
孙七 50
将上述信息按照成绩倒序输出

ScoreBean.java

public class ScoreBean implements WritableComparable<ScoreBean> {private String name;private int score;@Overridepublic int compareTo(ScoreBean o) {return o.getScore() - this.getScore();}@Overridepublic void write(DataOutput out) throws IOException {out.writeUTF(name);out.writeInt(score);}@Overridepublic void readFields(DataInput in) throws IOException {name = in.readUTF();score = in.readInt();}@Overridepublic String toString() {return this.name + "\t" + this.score;}//...省略getter和setter
}

Driver.java/Mapper.java/Reducer.java

public class ScoreDriver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(ScoreDriver.class);job.setMapperClass(Mapper.class);job.setReducerClass(Reducer.class);job.setMapOutputKeyClass(ScoreBean.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(ScoreBean.class);job.setOutputValueClass(NullWritable.class);FileInputFormat.setInputPaths(job,new Path("D:\\hadoop-study\\scoresort\\input"));FileOutputFormat.setOutputPath(job,new Path("D:\\hadoop-study\\scoresort\\output"));boolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}private static class Mapper extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text,ScoreBean,NullWritable> {private ScoreBean scoreBean = new ScoreBean();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String oneLine = value.toString();String[] split = oneLine.split(" ");scoreBean.setName(split[0]);scoreBean.setScore(Integer.valueOf(split[1]));context.write(scoreBean,NullWritable.get());}}private static class Reducer extends org.apache.hadoop.mapreduce.Reducer<ScoreBean,NullWritable,ScoreBean,NullWritable> {@Overrideprotected void reduce(ScoreBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {context.write(key,NullWritable.get());}}
}

自定义分组

分组我的理解是进入Reduce阶段前，将（逻辑）相同Key分为一组，迭代至reduce进行处理
重点：因为shuffle过程保证了map到reduce的有序性，所以在使用自定义分组时，首先要保证key排序力度包含分组。我理解的是需要在key排序中优先进行分组逻辑排序，其次考虑其他Key排序规则。

案例
格式大订单号\t商品名称\t销售价格
001 小米手机    1999
001 红米手机    999
001 华为手机    4999
002 三只松鼠    108
002 良品铺子    88
002 洗发水 52
003 MAC笔记本  10999
003 小米笔记本   3999
003 华为笔记本   5999
004 小米投影仪   2999
题目：找出每组订单中销售金额最高的两条记录
解题思路：（强烈说明：这里只是为了使用技术点来进行案例解刨，解题思路不止一种）
在map阶段首先对订单进行二次排序：订单号正序、销售金额倒叙
在reduce阶段通过自定义分组将相同订单号的订单分到一组，然后分组内输出前2个订单信息即可

Comparator.java
继承自org.apache.hadoop.io.WritableComparator
重写public int compare(WritableComparable a, WritableComparable b)方法，注意不要选择错误

package com.xbz.study.bigdata.hadoop.mr.groupCompator;import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;public class OrderGroupComparator extends WritableComparator{protected OrderGroupComparator() {//重点：必须，否则运行异常super(OrderBean.class,true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {OrderBean oa = (OrderBean) a;OrderBean ob = (OrderBean) b;//订单号相同即为同一组return oa.getOrderNo().compareTo(ob.getOrderNo());}
}

OrderBean.Java

package com.xbz.study.bigdata.hadoop.mr.groupCompator;import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class OrderBean implements WritableComparable<OrderBean> {private String orderNo;private String productName;private Float saleAmount;@Overridepublic int compareTo(OrderBean o) {//首先按照订单排序int firstRes = this.getOrderNo().compareTo(o.getOrderNo());//订单号相同，在按照销售金额倒序if(firstRes == 0){return o.getSaleAmount().compareTo(this.getSaleAmount());}return firstRes;}@Overridepublic String toString() {return this.orderNo + "\t" + this.productName + "\t" + this.saleAmount;}@Overridepublic void write(DataOutput out) throws IOException {out.writeUTF(orderNo);out.writeUTF(productName);out.writeFloat(saleAmount);}@Overridepublic void readFields(DataInput in) throws IOException {orderNo = in.readUTF();productName = in.readUTF();saleAmount = in.readFloat();}//getter和setter省略
}

Mapper.java/Reducer.java/Driver.java

package com.xbz.study.bigdata.hadoop.mr.groupCompator;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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 java.io.IOException;
import java.util.Iterator;public class OrderTopNDriver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(OrderTopNDriver.class);job.setMapperClass(MyMapper.class);job.setReducerClass(MyReducer.class);job.setMapOutputKeyClass(OrderBean.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(OrderBean.class);job.setOutputValueClass(NullWritable.class);FileInputFormat.setInputPaths(job,new Path("d:\\hadoop-study\\orderTopN\\input"));FileOutputFormat.setOutputPath(job,new Path("d:\\hadoop-study\\orderTopN\\output"));//设置自定义分组job.setGroupingComparatorClass(OrderGroupComparator.class);boolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}private static class MyMapper extends Mapper<LongWritable, Text,OrderBean,NullWritable> {private OrderBean orderBean = new OrderBean();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String oneLine = value.toString();String[] split = oneLine.split("\\t");//这里值得提醒的是，在赋值时如果有空值一定要记录覆盖原始字段，因为orderBean是重复使用的！！！orderBean.setOrderNo(split[0]);orderBean.setProductName(split[1]);orderBean.setSaleAmount(Float.valueOf(split[2]));context.write(orderBean,NullWritable.get());}}private static class MyReducer extends Reducer<OrderBean,NullWritable,OrderBean,NullWritable> {@Overrideprotected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {Iterator<NullWritable> iterator = values.iterator();int i = 0;//每组只取前2条while(i<2 && iterator.hasNext()){NullWritable next = iterator.next();context.write(key,NullWritable.get());i++;}}}
}

重点强调

KeyBean中的排序规则应该是先考虑分组排序列，在考虑其他排序列

分组比较器中必须在构造器中调用super，如下

protected OrderGroupComparator() {//重点：必须，否则运行异常super(OrderBean.class,true);
}

注意以上所有事情都做好了，不要忘记在job中设置自定义分组！！！
```
  //设置自定义分组job.setGroupingComparatorClass(OrderGroupComparator.class);
```

自定义RecordWriter

将一下文本内容中包含“baidu.com”的内容输出到baidu.log，其他内容输出到other.log
http://www.baidu.com
http://www.sina.com
http://www.tencent.com
http://www.google.com
思路解析：
这里我们只需要使用TextFileInputFormat读取数据，不需要使用MR，然后使用自定义的OutputFormat输出数据即可
OutputFormat.java

package com.xbz.study.bigdata.hadoop.mr.outputformat;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class MyOutputFormat extends FileOutputFormat<LongWritable, Text> {@Overridepublic RecordWriter getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {MyRecordWriter myRecordWriter = new MyRecordWriter();myRecordWriter.init(job);return myRecordWriter;}
}

RecordWriter.java

package com.xbz.study.bigdata.hadoop.mr.outputformat;import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class MyRecordWriter extends RecordWriter<LongWritable, Text> {private FSDataOutputStream baiduOut;private FSDataOutputStream otherOut;/*** 初始化，准备工作*/public void init(TaskAttemptContext job) throws IOException {//获取输出目录String outDir = job.getConfiguration().get(FileOutputFormat.OUTDIR);Path outDirPath = new Path(outDir);FileSystem fileSystem = FileSystem.get(job.getConfiguration());//创建输出流baiduOut = fileSystem.create(new Path(outDirPath, "baidu.log"));otherOut = fileSystem.create(new Path(outDirPath,"other.log"));}@Overridepublic void write(LongWritable key, Text value) throws IOException, InterruptedException {String oneLine = value.toString();if(oneLine.contains("baidu")){baiduOut.write((oneLine+"\n").getBytes());}else{otherOut.write((oneLine+"\n").getBytes());}}@Overridepublic void close(TaskAttemptContext context) throws IOException, InterruptedException {IOUtils.closeStream(baiduOut);IOUtils.closeStream(otherOut);}
}

Driver.java

package com.xbz.study.bigdata.hadoop.mr.outputformat;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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 Driver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(Driver.class);job.setOutputKeyClass(LongWritable.class);job.setOutputValueClass(Text.class);FileInputFormat.setInputPaths(job,new Path("D:\\hadoop-study\\outputformat\\input"));FileOutputFormat.setOutputPath(job,new Path("D:\\hadoop-study\\outputformat\\output"));job.setOutputFormatClass(MyOutputFormat.class);boolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}
}

大数据入门--hadoop（三）--MR编程相关推荐

大数据入门笔记(三)
为什么80%的码农都做不了架构师?>>> HDFS( Hadoop Distributed File System) hadoop分布式文件系统总的设计思想: 分而治之 -- ...
2018今日头条校招大数据方向第三批编程题2
链接:https://www.nowcoder.com/questionTerminal/1f8d61e3090644d8996fdec01694d3cf 来源:牛客网有一个推箱子的游戏, 一开始的 ...
大数据入门（八）win10下的wordcount
目录方法一上传文件到hdfs java project 方法二参考有两种方法:方法一需要借用eclipse自己编写代码,优点是有助于理解mapreduce,缺点复杂.方法二可以直接调用Hado ...
大数据入门-三分钟读懂Hadoop
最近在收集整理大数据入门文章,各位盆友关注点赞不迷路,每天都要开心鸭! 大数据入门系列文章 1.大数据入门-大数据是什么 1.大数据入门-大数据是什么 2.大数据入门-大数据技术概述(一) 2.大数据 ...
大数据之Hadoop入门
1 大数据概论大数据(Big Data):指无法在一定时间范围内用常规软件工具进行捕捉.管理和处理的数据集合,是需要新处理模式才能具有更强决策力.洞察发现力和流程优化能力的海量.高增长率和多样化的信 ...
大数据之Hadoop学习——动手实战学习MapReduce编程实例
文章目录一.MapReduce理论基础二.Hadoop.Spark学习路线及资源收纳三.MapReduce编程实例 1.自定义对象序列化需求分析报错:Exception in thread ...
大数据技术Hadoop入门理论系列之一----hadoop生态圈介绍
Technorati 标记: hadoop,生态圈,ecosystem,yarn,spark,入门 1. hadoop 生态概况 Hadoop是一个由Apache基金会所开发的分布式系统基础架构. 用 ...
【大数据】Hadoop （一）入门
课程链接: 尚硅谷大数据Hadoop 3.x(入门搭建+安装调优)_哔哩哔哩_bilibili (Hadoop到底是干什么用的? - 知乎 (zhihu.com) 资料 Hadoop中文文档 (apa ...
华为云大数据存储的冗余方式是三副本_大数据入门：HDFS数据副本存放策略
大数据处理当中,数据储存始终是一个重要的环节,从现阶段的市场现状来说,以Hadoop为首的大数据技术框架,仍然占据主流地位,而Hadoop的HDFS,在数据存储方面,仍然得到重用.今天的大数据入门分享 ...

大数据入门--hadoop（三）--MR编程