大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL)+Hadoop数据压缩+Yarn资源调度器
5.6 任务的推测执行（秘籍）
3.3.4 WritableComparable排序
0、排序概述
1、排序的分类
2、自定义排序WritableComparable
（1）原理分析
bean对象做为key传输，需要实现WritableComparable接口重写compareTo()方法，就可以实现排序。
@Override
public int compareTo(FlowBean o) {
int result;
// 按照总流量大小，倒序排列
if (sumFlow > bean.getSumFlow()) {
result = -1;
} else if (sumFlow < bean.getSumFlow()) {
result = 1;
} else {
result = 0;
}
return result;
}
3.3.5 WritableComparable排序案例实操（全排序）
1、需求
根据案例2.3产生的结果再次对总流量进行排序。
（1）输入数据
（2）期望输出数据
13509468723 7335 110349 117684
13736230513 2481 24681 27162
13956435636 132 1512 1644
13846544121 264 0 264
......
2、需求分析
3、代码实现
（1）FlowBean对象在在需求1基础上增加了比较功能
package com.atguigu.mr.sort;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
public class FlowBean implements WritableComparable<FlowBean> {
private long upFlow; // 上行流量
private long downFlow; // 下行流量
private long sumFlow; // 总流量
/**
* 反序列化时，需要反射调用空参构造函数，所以必须有
*/
public FlowBean() {
super();
}
public FlowBean(long upFlow, long downFlow) {
super();
this.upFlow = upFlow;
this.downFlow = downFlow;
this.sumFlow = upFlow + downFlow;
}
/**
* 序列化方法
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeLong(upFlow);
out.writeLong(downFlow);
out.writeLong(sumFlow);
}
/**
* 反序列化方法，注意反序列化的顺序和序列化的顺序完全一致
*/
@Override
public void readFields(DataInput in) throws IOException {
upFlow = in.readLong();
downFlow = in.readLong();
sumFlow = in.readLong();
}
/**
* 比较方法
*/
@Override
public int compareTo(FlowBean bean) {
int result;
// 按照总流量大小，倒序排列
if (sumFlow > bean.getSumFlow()) {
result = -1;
} else if (sumFlow < bean.getSumFlow()) {
result = 1;
} else {
result = 0;
}
return result;
}
public long getUpFlow() {
return upFlow;
}
public void setUpFlow(long upFlow) {
this.upFlow = upFlow;
}
public long getDownFlow() {
return downFlow;
}
public void setDownFlow(long downFlow) {
this.downFlow = downFlow;
}
public long getSumFlow() {
return sumFlow;
}
public void setSumFlow(long sumFlow) {
this.sumFlow = sumFlow;
}
@Override
public String toString() {
return upFlow + " " + downFlow + " " + sumFlow;
}
}
（2）编写Mapper类
package com.atguigu.mr.sort;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {
FlowBean k = new FlowBean();
Text v = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 13736230513 2481 24681 27162
// 1、获取一行
String line = value.toString();
// 2、截取
String[] fields = line.split(" ");
// 3、封装对象
String phoneNum = fields[0];
long upFlow = Long.parseLong(fields[1]);
long downFlow = Long.parseLong(fields[2]);
long sumFlow = Long.parseLong(fields[3]);
k.setUpFlow(upFlow);
k.setDownFlow(downFlow);
k.setSumFlow(sumFlow);
v.set(phoneNum);
// 4、输出
context.write(k, v);
}
}
（3）编写Reducer类
package com.atguigu.mr.sort;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean> {
@Override
protected void reduce(FlowBean key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
// 循环输出，避免总流量相同的情况
for (Text text : values) {
context.write(text, key);
}
}
}
（4）编写Driver类
package com.atguigu.mr.sort;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 FlowCountSortDriver {
public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "d:/temp/atguigu/0529/output2", "d:/temp/atguigu/0529/output8" };
// 1、获取配置信息，或者job对象实例
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);
// 2、指定本程序的jar包所在的本地路径
job.setJarByClass(FlowCountSortDriver.class);
// 3、指定本业务job要使用的mapper/Reducer业务类
job.setMapperClass(FlowCountSortMapper.class);
job.setReducerClass(FlowCountSortReducer.class);
// 4、指定mapper输出数据的kv类型
job.setMapOutputKeyClass(FlowBean.class);
job.setMapOutputValueClass(Text.class);
// 5、指定最终输出的数据的kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
// 6、指定job的输入原始文件所在目录
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7、将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
3.3.6 WritableComparable排序案例实操（区内排序）
1、需求
要求每个省份手机号输出的文件中按照总流量内部排序。
2、需求分析
基于前一个需求，增加自定义分区类，分区按照省份手机号设置。
3、案例实操
（1）增加自定义分区类
package com.atguigu.mr.sort;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class ProvincePartitioner extends Partitioner<FlowBean, Text> {
@Override
public int getPartition(FlowBean key, Text value, int numPartitions) {
// 按照手机号的前三位进行分区
// 获取手机号的前三位
String prePhoneNum = value.toString().substring(0, 3);
// 根据手机号归属地设置分区
int partition = 4;
if ("136".equals(prePhoneNum)) {
partition = 0;
} else if ("137".equals(prePhoneNum)) {
partition = 1;
} else if ("138".equals(prePhoneNum)) {
partition = 2;
} else if ("139".equals(prePhoneNum)) {
partition = 3;
}
return partition;
}
}
（2）在驱动类中添加加载分区类
// 加载自定义分区类(即关联分区)
job.setPartitionerClass(ProvincePartitioner.class);
// 设置Reducetask个数
job.setNumReduceTasks(5);
3.3.7 Combiner合并
Combiner合并是Hadoop框架优化的一种手段，因为Combiner合并减少了数据的IO传输。
（6）自定义Combiner实现步骤
（a）自定义一个Combiner继承Reducer，重写reduce()方法
package com.atguigu.mr.wordcount;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
int sum;
IntWritable v = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
// 1、汇总，累加求和
sum = 0;
for (IntWritable value : values) {
sum += value.get();
}
v.set(sum);
// 2、写出
context.write(key, v);
}
}
（b）在Job驱动类中设置：
job.setCombinerClass(WordcountCombiner.class);
3.3.8 Combiner合并案例实操
1、需求
统计过程中对每一个MapTask的输出进行局部汇总，以减小网络传输量即采用Combiner功能。
（1）数据输入
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
（2）期望输出数据
期望：Combine输入数据多，输出时经过合并，输出数据降低。
2、需求分析
3、案例实操-方案一
1）增加一个WordcountCombiner类继承Reducer
package com.atguigu.mr.wordcount;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
int sum;
IntWritable v = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
// 1、汇总，累加求和
sum = 0;
for (IntWritable value : values) {
sum += value.get();
}
v.set(sum);
// 2、写出
context.write(key, v);
}
}
2）在WordcountDriver驱动类中指定Combiner
// 指定需要使用Combiner，以及用哪个类作为Combiner的逻辑
job.setCombinerClass(WordcountCombiner.class);
4、案例实操-方案二
1）将WordcountReducer作为Combiner在WordcountDriver驱动类中指定
// 指定需要使用Combiner，以及用哪个类作为Combiner的逻辑
job.setCombinerClass(WordcountReducer.class);
运行程序，如下图所示：
3.3.9 GroupingComparator分组（辅助排序/分组排序）
对Reduce阶段的数据根据某一个或几个字段进行分组。
分组排序步骤：
（1）自定义类继承WritableComparator
（2）重写compare()方法
@Override
public int compare(WritableComparable a, WritableComparable b) {
// 比较的业务逻辑
// ......
return result;
}
（3）创建一个构造将比较对象的类传给父类
protected OrderGroupingComparator() {
super(OrderBean.class, true);
}
3.3.10 GroupingComparator分组案例实操
1、需求
有如下订单数据
现在需要求出每一个订单中最贵的商品。
（1）输入数据
GroupingComparator.txt
0000001 Pdt_01 222.8
0000002 Pdt_05 722.4
0000001 Pdt_02 33.8
0000003 Pdt_06 232.8
0000003 Pdt_02 33.8
0000002 Pdt_03 522.8
0000002 Pdt_04 122.4
（2）期望输出数据
1 222.8
2 722.4
3 232.8
2、需求分析
（1）利用“订单id和成交金额”作为key，可以将Map阶段读取到的所有订单数据按照id升序排序，如果id相同再按照金额降序排序，发送到Reduce。
（2）在Reduce端利用GroupingComparator将订单id相同的kv聚合成组，然后取第一个即是该订单中最贵商品，如下图所示。
3、代码实现
（1）定义订单信息OrderBean类
package com.atguigu.mr.order;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
public class OrderBean implements WritableComparable<OrderBean> {
private int order_id; // 订单id
private double price; // 订单价格
public OrderBean() {
super();
}
public OrderBean(int order_id, double price) {
super();
this.order_id = order_id;
this.price = price;
}
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(order_id);
out.writeDouble(price);
}
@Override
public void readFields(DataInput in) throws IOException {
this.order_id = in.readInt();
this.price = in.readDouble();
}
// 二次排序
@Override
public int compareTo(OrderBean bean) {
// 先按照订单id升序排序，如果订单id相同则按照价格降序排序
int result;
if (order_id > bean.getOrder_id()) {
result = 1;
} else if (order_id < bean.getOrder_id()) {
result = -1;
} else {
if (price > bean.getPrice()) {
result = -1;
} else if (price < bean.getPrice()) {
result = 1;
} else {
result = 0;
}
}
return result;
}
public int getOrder_id() {
return order_id;
}
public void setOrder_id(int order_id) {
this.order_id = order_id;
}
public double getPrice() {
return price;
}
public void setPrice(double price) {
this.price = price;
}
@Override
public String toString() {
return order_id + " " + price;
}
}
（2）编写OrderSortMapper类
package com.atguigu.mr.order;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
// 0000001 Pdt_01 222.8
public class OrderSortMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {
OrderBean k = new OrderBean();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 0000001 Pdt_01 222.8
// 1、获取一行
String line = value.toString();
// 2、截取
String[] fields = line.split(" ");
// 3、封装对象
k.setOrder_id(Integer.parseInt(fields[0]));
k.setPrice(Double.parseDouble(fields[2]));
// 4、写出
context.write(k, NullWritable.get());
}
}
（3）编写OrderSortGroupingComparator类
package com.atguigu.mr.order;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class OrderSortGroupingComparator extends WritableComparator {
/**
* 创建一个构造将比较对象的类传给父类
*/
protected OrderSortGroupingComparator() {
super(OrderBean.class, true);
}
@SuppressWarnings("rawtypes")
@Override
public int compare(WritableComparable a, WritableComparable b) {
// 比较的业务逻辑
// 要求按照只要是id相同，就认为是相同的key
OrderBean aBean = (OrderBean) a;
OrderBean bBean = (OrderBean) b;
int result;
if (aBean.getOrder_id() > bBean.getOrder_id()) {
result = 1;
} else if (aBean.getOrder_id() < bBean.getOrder_id()) {
result = -1;
} else {
result = 0;
}
return result;
}
}
（4）编写OrderSortReducer类
package com.atguigu.mr.order;
import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
public class OrderSortReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable>{
@Override
protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context)
throws IOException, InterruptedException {
// 只写出第一行数据
context.write(key, NullWritable.get());
// 出每个reduce的所有数据
/*
for (NullWritable nullWritable : values) {
context.write(key, NullWritable.get());
}
*/
}
}
（5）编写OrderSortDriver类
package com.atguigu.mr.order;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class OrderSortDriver {
public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "d:/temp/atguigu/0529/input/inputorder", "d:/temp/atguigu/0529/output11" };
// 1、获取配置信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 2、设置jar包加载路径
job.setJarByClass(OrderSortDriver.class);
// 3、加载map/reduce类
job.setMapperClass(OrderSortMapper.class);
job.setReducerClass(OrderSortReducer.class);
// 4、设置map输出数据key和value类型
job.setMapOutputKeyClass(OrderBean.class);
job.setMapOutputValueClass(NullWritable.class);
// 5、设置最终输出数据的key和value类型
job.setOutputKeyClass(OrderBean.class);
job.setOutputValueClass(NullWritable.class);
// 6、设置输入数据和输出数据路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 8、设置reduce端的分组
job.setGroupingComparatorClass(OrderSortGroupingComparator.class);
// 7、提交
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
3.4 MapTask工作机制
MapTask工作机制如下图所示。
- （1）Read阶段：MapTask通过用户编写的RecordReader，从输入InputSplit中解析出一个个key/value。
- （2）Map阶段：该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。
- （3）Collect收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。
- （4）
Spill阶段：即“溢写”，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。 溢写阶段详情：- 步骤1：利用`快速排序算法`对缓存区内的数据进行排序，排序方式是，`先按照分区编号Partition进行排序，然后按照key进行排序`。这样，经过排序后，`数据以分区为单位聚集在一起`，且`同一分区内所有数据按照key有序`。
- 步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件`output/spillN.out`（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次`聚集`操作。
- 步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件`output/spillN.out.index`中。
- （5）Combine阶段：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。
当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。
在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。
让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。
3.5 ReduceTask工作机制
1、ReduceTask工作机制
ReduceTask工作机制，如下图所示。
- （1）Copy阶段：ReduceTask从各个MapTask上远程
拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则溢写到磁盘上，否则直接放到内存中。 - （2）Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个
后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。 - （3）Sort阶段：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了
将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。 - （4）Reduce阶段：reduce()函数将计算结果写到HDFS上。
2、设置ReduceTask并行度（个数）
ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：
// 默认值是1，手动设置为4
job.setNumReduceTasks(4);
3、实验：测试ReduceTask多少合适
（1）实验环境：1个Master节点，16个Slave节点：CPU:8GHZ，内存：2G，数据量：1GB，在不考虑分区的情况下
（2）实验结论：
4、注意事项
3.6 OutputFormat数据输出
3.6.1 OutputFormat接口实现类
3.6.2 自定义OutputFormat
3.6.3 自定义OutputFormat案例实操
1、需求
过滤输入的log日志，包含atguigu的网站输出到e:/atguigu.log，不包含atguigu的网站输出到e:/other.log。
（1）输入数据
log.txt
http://www.baidu.com
http://www.google.com
http://cn.bing.com
http://www.atguigu.com
http://www.sohu.com
http://www.sina.com
http://www.sin2a.com
http://www.sin2desa.com
http://www.sindsafa.com
（2）期望输出数据
2、需求分析
3、案例实操
（1）编写FilterMapper类
package com.atguigu.mr.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class FilterMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 写出
context.write(value, NullWritable.get());
}
}
（2）编写FilterReducer类
package com.atguigu.mr.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class FilterReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
Text k = new Text();
@Override
protected void reduce(Text key, Iterable<NullWritable> values, Context context)
throws IOException, InterruptedException {
// 1、获取一行
String line = key.toString();
// 2、拼接
line = line + "
";
// 3、设置key
k.set(line);
// 4、输出
context.write(k, NullWritable.get());
}
}
（3）自定义一个OutputFormat类
package com.atguigu.mr.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.NullWritable;
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;
public class FilterOutputFormat extends FileOutputFormat<Text, NullWritable> {
@Override
public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job)
throws IOException, InterruptedException {
// 创建一个类FRecordWriter继承RecordWriter
return new FRecordWriter(job);
}
}
（4）编写RecordWriter类
package com.atguigu.mr.outputformat;
import java.io.IOException;
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.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
public class FRecordWriter extends RecordWriter<Text, NullWritable> {
FSDataOutputStream atguiguOut = null;
FSDataOutputStream otherOut = null;
public FRecordWriter(TaskAttemptContext job) {
// 1、获取文件系统
FileSystem fs;
try {
fs = FileSystem.get(job.getConfiguration());
// 2、创建输出文件路径
Path atguiguPath = new Path("d:/temp/atguigu/0529/atguigu.log");
Path otherPath = new Path("d:/temp/atguigu/0529/other.log");
// 3、创建输出流
atguiguOut = fs.create(atguiguPath);
otherOut = fs.create(otherPath);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void write(Text key, NullWritable value) throws IOException, InterruptedException {
// 判断key当中是否包含“atguigu”，输出到不同的文件
if (key.toString().contains("atguigu")) {
atguiguOut.write(key.toString().getBytes());
} else {
otherOut.write(key.toString().getBytes());
}
}
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
// 关闭资源
IOUtils.closeStream(atguiguOut);
IOUtils.closeStream(otherOut);
}
}
（5）编写FilterDriver类
package com.atguigu.mr.outputformat;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class FilterDriver {
public static void main(String[] args)
throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "d:/temp/atguigu/0529/input/inputoutputformat", "d:/temp/atguigu/0529/output12" };
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(FilterDriver.class);
job.setMapperClass(FilterMapper.class);
job.setReducerClass(FilterReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 要将自定义的输出格式组件设置到job中
job.setOutputFormatClass(FilterOutputFormat.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
// 虽然我们自定义了outputformat，但是因为我们的outputformat继承自fileoutputformat
// 而fileoutputformat要输出一个_SUCCESS文件(标记文件)，所以，在这还得指定一个输出目录
FileOutputFormat.setOutputPath(job, new Path(args[1]));
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
3.7 Join多种应用
3.7.1 Reduce Join
3.7.2 Reduce Join案例实操
1、需求
2、需求分析
通过将关联条件作为Map输出的key，将两表满足Join条件的数据并携带数据所来源的文件信息，发往同一个ReduceTask，在Reduce中进行数据的串联，如下图所示。（
注意：需求分析中的数据仅仅是一个示例）3、代码实现
1）创建商品和订合并后的Bean类
package com.atguigu.mr.table;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Writable;
public class TableBean implements Writable {
// id pid amount
// pid pname
private String id; // 订单id
private String pid; // 产品id
private int amount; // 产品数量
private String pname; // 产品名称
private String flag; // 定义一个标记，用于标记是订单表还是产品表
public TableBean() {
super();
}
public TableBean(String id, String pid, int amount, String pname, String flag) {
super();
this.id = id;
this.pid = pid;
this.amount = amount;
this.pname = pname;
this.flag = flag;
}
/**
* 序列化方法
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(id);
out.writeUTF(pid);
out.writeInt(amount);
out.writeUTF(pname);
out.writeUTF(flag);
}
/**
* 反序列化方法
*/
@Override
public void readFields(DataInput in) throws IOException {
this.id = in.readUTF();
this.pid = in.readUTF();
this.amount = in.readInt();
this.pname = in.readUTF();
this.flag = in.readUTF();
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public int getAmount() {
return amount;
}
public void setAmount(int amount) {
this.amount = amount;
}
public String getPname() {
return pname;
}
public void setPname(String pname) {
this.pname = pname;
}
public String getFlag() {
return flag;
}
public void setFlag(String flag) {
this.flag = flag;
}
@Override
public String toString() {
return id + " " + pname + " " + amount + " " ;
}
}
2）编写TableMapper类
package com.atguigu.mr.table;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {
String name;
TableBean bean = new TableBean();
Text k = new Text();
@Override
protected void setup(Context context)
throws IOException, InterruptedException {
// 1、获取输入文件的切片
FileSplit split = (FileSplit) context.getInputSplit();
// 2、获取输入文件的名称
name = split.getPath().getName();
}
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// id pid amount
// 1001 01 1
// pid pname
// 01 小米
// 1、获取输入数据（一行数据）
String line = value.toString();
// 2、不同文件分别处理
if (name.startsWith("order")) {
// 2.1、切割
String[] fields = line.split(" ");
// 2.2、封装bean对象
bean.setId(fields[0]);
bean.setPid(fields[1]);
bean.setAmount(Integer.parseInt(fields[2]));
bean.setPname("");
bean.setFlag("order");
k.set(fields[1]);
} else {
String[] fields = line.split(" ");
bean.setId("");
bean.setPid(fields[0]);
bean.setAmount(0);
bean.setPname(fields[1]);
bean.setFlag("pd");
k.set(fields[0]);
}
// 3、写出
context.write(k, bean);
}
}
3）编写TableReducer类
package com.atguigu.mr.table;
import java.io.IOException;
import java.util.ArrayList;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {
@Override
protected void reduce(Text key, Iterable<TableBean> values,
Context context) throws IOException, InterruptedException {
// 1、准备存储订单的集合
ArrayList<TableBean> orderBeans = new ArrayList<>();
// 2、准备存储产品的Bean（一个产品对应多个订单）
TableBean pdBean = new TableBean();
// 3、遍历所有的Bean
for (TableBean tableBean : values) {
if ("order".equals(tableBean.getFlag())) { // 说明是订单表
// 拷贝传递过来的每条订单数据到集合中，因为values存放的是对象的引用(地址值)，而不是对象本身，即java中只有值传递
TableBean orderBean = new TableBean();
try {
BeanUtils.copyProperties(orderBean, tableBean);
orderBeans.add(orderBean);
} catch (Exception e) {
e.printStackTrace();
}
} else { // 说明是产品表
try {
BeanUtils.copyProperties(pdBean, tableBean);
} catch (Exception e) {
e.printStackTrace();
}
}
}
// 4、表的拼接
for (TableBean orderBean : orderBeans) {
orderBean.setPname(pdBean.getPname());
// 5、数据写出去
context.write(orderBean, NullWritable.get());
}
}
}
4）编写TableDriver类
package com.atguigu.mr.table;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class TableDriver {
public static void main(String[] args)
throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
// 0、根据自己电脑路径重新配置
args = new String[] { "d:/temp/atguigu/0529/input/inputtable", "d:/temp/atguigu/0529/output13" };
// 1、获取配置信息，或者job对象实例
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);
// 2、指定本程序的jar包所在的本地路径
job.setJarByClass(TableDriver.class);
// 3、指定本业务job要使用的Mapper/Reducer业务类
job.setMapperClass(TableMapper.class);
job.setReducerClass(TableReducer.class);
// 4、指定Mapper输出数据的kv类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(TableBean.class);
// 5、指定最终输出的数据的kv类型
job.setOutputKeyClass(TableBean.class);
job.setOutputValueClass(NullWritable.class);
// 6、指定job的输入原始文件所在目录
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7、将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
4、测试
运行程序查看结果
1004 小米 4
1001 小米 1
1005 华为 5
1002 华为 2
1006 格力 6
1003 格力 3
5、总结
3.7.3 Map Join
1、使用场景
Map Join 适用于一张表十分小、一张表很大(在内存中存不下)的场景。
2、优点
思考：在Reduce端处理过多的表，非常容易产生数据倾斜。怎么办？
在Map端缓存多张表，提前处理业务逻辑，这样增加Map端业务，减少Reduce端数据的压力，尽可能的减少数据倾斜。
3、具体办法：采用DistributedCache
（1）在Mapper的setup阶段，将文件读取到缓存集合中。
（2）在驱动函数中加载缓存。
// 缓存普通文件到Task运行节点。
job.addCacheFile(new URI("file://e:/cache/pd.txt"));
3.7.4 Map Join案例实操
1、需求
2、需求分析
MapJoin适用于关联表中有小表的情形。
3、实现代码
（1）先在驱动模块中添加缓存文件
package com.atguigu.mr.cache;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class DistributedCacheDriver {
public static void main(String[] args) throws Exception {
// 0、根据自己电脑路径重新配置
args = new String[] { "d:/temp/atguigu/0529/input/inputtable2", "d:/temp/atguigu/0529/output14" };
// 1、获取job信息
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);
// 2、设置加载jar包路径
job.setJarByClass(DistributedCacheDriver.class);
// 3、关联map
job.setMapperClass(DistributedCacheMapper.class);
// 4、设置最终输出数据类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 5、设置输入输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 6、加载缓存数据(文件)
job.addCacheFile(new URI("file:///d:/temp/atguigu/0529/input/inputcache/pd.txt"));
// 7、Map端Join的逻辑不需要Reduce阶段，设置ReduceTask数量为0
job.setNumReduceTasks(0);
// 8、提交
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
（2）读取并处理缓存的文件数据
package com.atguigu.mr.cache;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
* 缓存小表
* @author bruce
*/
public class DistributedCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
HashMap<String, String> pdMap = new HashMap<>();
@Override
protected void setup(Context context)
throws IOException, InterruptedException {
// 1、获取缓存的文件
URI[] cacheFiles = context.getCacheFiles();
String path = cacheFiles[0].getPath().toString();
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(path), "UTF-8"));
String line;
while(StringUtils.isNotEmpty(line = bufferedReader.readLine())) {
// pid pname
// 01 小米
// 2、切割
String[] fields = line.split(" ");
// 3、缓存数据到集合
pdMap.put(fields[0], fields[1]);
}
// 4、关流
bufferedReader.close();
}
Text k = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// id pid amount
// 1001 01 1
// pid pname
// 01 小米
// 1、获取一行
String line = value.toString();
// 2、截取
String[] fields = line.split(" ");
// 3、获取产品id
String pid = fields[1];
// 4、获取商品名称
String pname = pdMap.get(pid);
// 5、拼接
k.set(line + " " + pname);
// 6、写出
context.write(k, NullWritable.get());
}
}
3.8 计数器应用
3.9 数据清洗（ETL）
在运行核心业务MapReduce程序之前，往往要先对数据进行清洗，清理掉不符合用户要求的数据。清理的过程往往只需要运行Mapper程序，不需要运行Reduce程序。
3.9.1 数据清洗案例实操-简单解析版
1、需求
去除日志中字段长度小于等于11的日志。
（1）输入数据
（2）期望输出数据
每行字段长度都大于11。
2、需求分析
需要在Map阶段对输入的数据根据规则进行过滤清洗。
3、实现代码
（1）编写LogMapper类
package com.atguigu.mr.log;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
Text k = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 1、获取一行数据
String line = value.toString();
// 2、解析数据
boolean result = parseLog(line, context);
// 3、说明解析失败，则退出
if (!result) {
return;
}
// 4、设置key
k.set(line);
// 5、说明解析成功，写出数据
context.write(k, NullWritable.get());
}
/**
* 解析数据的方法
* @param line
* @param context
* @return
*/
private boolean parseLog(String line, Context context) {
// 1、截取
String[] fields = line.split(" ");
// 2、判断日志长度大于11的为合法，演示阶段，只有一个判断条件，实际开发中有很多个判断条件
if (fields.length > 11) {
// 使用系统计数器
context.getCounter("map", "true").increment(1);;
return true;
} else {
context.getCounter("map", "false").increment(1);
return false;
}
}
}
（2）编写LogDriver类
package com.atguigu.mr.log;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class LogDriver {
public static void main(String[] args)
throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
// 0、输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "d:/temp/atguigu/0529/input/inputlog", "d:/temp/atguigu/0529/output15" };
// 1、获取job信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 2、加载jar包
job.setJarByClass(LogDriver.class);
// 3、关联map
job.setMapperClass(LogMapper.class);
// 4、设置最终输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 设置reducetask个数为0
job.setNumReduceTasks(0);
// 5、设置输入和输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 6、提交
job.waitForCompletion(true);
}
}
3.9.2 数据清洗案例实操-复杂解析版
1、需求
对Web访问日志中的各字段识别切分，去除日志中不合法的记录。根据清洗规则，输出过滤后的数据。
（1）输入数据
（2）期望输出数据
都是合法的数据。
2、实现代码
（1）定义一个bean，用来记录日志数据中的各数据字段
package com.atguigu.mr.log2;
public class LogBean {
private String remote_addr; // 记录客户端的ip地址
private String remote_user; // 记录客户端用户名称，忽略属性"-"
private String time_local; // 记录访问时间与时区
private String request; // 记录请求的url与http协议
private String status; // 记录请求状态；成功是200
private String body_bytes_sent; // 记录发送给客户端文件主体内容大小
private String http_referer; // 用来记录从那个页面链接访问过来的
private String http_user_agent; // 记录客户浏览器的相关信息
private boolean valid = true; // 判断数据是否合法
public String getRemote_addr() {
return remote_addr;
}
public void setRemote_addr(String remote_addr) {
this.remote_addr = remote_addr;
}
public String getRemote_user() {
return remote_user;
}
public void setRemote_user(String remote_user) {
this.remote_user = remote_user;
}
public String getTime_local() {
return time_local;
}
public void setTime_local(String time_local) {
this.time_local = time_local;
}
public String getRequest() {
return request;
}
public void setRequest(String request) {
this.request = request;
}
public String getStatus() {
return status;
}
public void setStatus(String status) {
this.status = status;
}
public String getBody_bytes_sent() {
return body_bytes_sent;
}
public void setBody_bytes_sent(String body_bytes_sent) {
this.body_bytes_sent = body_bytes_sent;
}
public String getHttp_referer() {
return http_referer;
}
public void setHttp_referer(String http_referer) {
this.http_referer = http_referer;
}
public String getHttp_user_agent() {
return http_user_agent;
}
public void setHttp_user_agent(String http_user_agent) {
this.http_user_agent = http_user_agent;
}
public boolean isValid() {
return valid;
}
public void setValid(boolean valid) {
this.valid = valid;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(this.valid);
sb.append(" 01").append(this.remote_addr);
sb.append(" 01").append(this.remote_user);
sb.append(" 01").append(this.time_local);
sb.append(" 01").append(this.request);
sb.append(" 01").append(this.status);
sb.append(" 01").append(this.body_bytes_sent);
sb.append(" 01").append(this.http_referer);
sb.append(" 01").append(this.http_user_agent);
return sb.toString();
}
}
（2）编写LogMapper类
package com.atguigu.mr.log2;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
Text k = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 1、获取1行
String line = value.toString();
// 2、解析日志是否合法
LogBean bean = parseLog(line);
if (!bean.isValid()) {
return;
}
k.set(bean.toString());
// 3、输出
context.write(k, NullWritable.get());
}
/**
* 解析日志的方法
* @param line
* @return
*/
private LogBean parseLog(String line) {
LogBean logBean = new LogBean();
// 1、截取
String[] fields = line.split(" ");
if (fields.length > 11) {
// 2、封装数据
logBean.setRemote_addr(fields[0]);
logBean.setRemote_user(fields[1]);
logBean.setTime_local(fields[3].substring(1));
logBean.setRequest(fields[6]);
logBean.setStatus(fields[8]);
logBean.setBody_bytes_sent(fields[9]);
logBean.setHttp_referer(fields[10]);
if (fields.length > 12) {
logBean.setHttp_user_agent(fields[11] + " " + fields[12]);
} else {
logBean.setHttp_user_agent(fields[11]);
}
// 状态码大于400，HTTP错误
if (Integer.parseInt(logBean.getStatus()) >= 400) {
logBean.setValid(false);
}
} else {
logBean.setValid(false);
}
return logBean;
}
}
（3）编写LogDriver类
package com.atguigu.mr.log2;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class LogDriver {
public static void main(String[] args)
throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
// 0、输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "d:/temp/atguigu/0529/input/inputlog", "d:/temp/atguigu/0529/output16" };
// 1、获取job信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 2、加载jar包
job.setJarByClass(LogDriver.class);
// 3、关联map
job.setMapperClass(LogMapper.class);
// 4、设置最终输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 设置reducetask个数为0
job.setNumReduceTasks(0);
// 5、设置输入和输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 6、提交
job.waitForCompletion(true);
}
}
3.10 MapReduce开发总结
在编写MapReduce程序时，需要考虑如下几个方面：
MapReduce开发总结01
MapReduce开发总结02
MapReduce开发总结03
MapReduce开发总结04
MapReduce开发总结05
第4章 Hadoop数据压缩
4.1 概述
压缩策略和原则
4.2 MR支持的压缩编码
为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器，如下表所示。
压缩性能的比较
http://google.github.io/snappy/
On a single core of a Core i7 processor in 64-bit mode, Snappy compresses at about 250 MB/sec or more and decompresses at about 500 MB/sec or more.
4.3 压缩方式选择
4.3.1 Gzip压缩
4.3.2 Bzip2压缩
4.3.3 Lzo压缩
4.3.4 Snappy压缩
4.4 压缩位置选择
压缩可以在MapReduce作用的任意阶段启用，如下图所示。
4.5 压缩参数配置
要在Hadoop中启用压缩，可以配置如下参数：
4.6 压缩实操案例
4.6.1 数据流的压缩和解压缩
测试一下如下压缩方式：
DEFLATE org.apache.hadoop.io.compress.DefaultCodec
gzip org.apache.hadoop.io.compress.GzipCodec
bzip2 org.apache.hadoop.io.compress.BZip2Codec
测试代码如下：
package com.atguigu.mr.compress;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
public class TestCompress {
public static void main(String[] args) throws ClassNotFoundException, IOException {
// 1、压缩
// compress("d:/temp/atguigu/0529/hello.txt", "org.apache.hadoop.io.compress.DefaultCodec");
// compress("d:/temp/atguigu/0529/hello.txt", "org.apache.hadoop.io.compress.GzipCodec");
compress("d:/temp/atguigu/0529/hello.txt", "org.apache.hadoop.io.compress.BZip2Codec");
// 2、解压缩
decompress("d:/temp/atguigu/0529/hello.txt.bz2");
// decompress("d:/temp/atguigu/0529/hello.txt.gz");
// decompress("d:/temp/atguigu/0529/hello.txt.deflate");
}
private static void compress(String filename, String method) throws ClassNotFoundException, IOException {
// （1）获取输入流
FileInputStream fis = new FileInputStream(new File(filename));
Class<?> codecClass = Class.forName(method); // 反射
CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, new Configuration());
// （2）获取输出流
FileOutputStream fos = new FileOutputStream(new File(filename + codec.getDefaultExtension()));
CompressionOutputStream cos = codec.createOutputStream(fos);
// （3）流的对拷
IOUtils.copyBytes(fis, cos, 1024 * 1024 * 5, false);
// （4）关闭资源
cos.close();
fos.close();
fis.close();
}
private static void decompress(String filename) throws FileNotFoundException, IOException {
// （0）校验是否能解压缩
CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
CompressionCodec codec = factory.getCodec(new Path(filename));
if (codec == null) {
System.out.println("cannot find codec for file " + filename);
}
// （1）获取输入流
FileInputStream fis = new FileInputStream(new File(filename));
CompressionInputStream cis = codec.createInputStream(fis);
// （2）获取输出流
FileOutputStream fos = new FileOutputStream(new File(filename + ".decoded"));
// （3）流的对拷
IOUtils.copyBytes(cis, fos, 1024 * 1024 * 5, false);
// （4）关闭资源
fos.close();
cis.close();
fis.close();
}
}
4.6.2 Map输出端采用压缩
即使你的MapReduce的输入输出文件都是未压缩的文件，你仍然可以对Map任务的中间结果输出做压缩，因为它要写在硬盘并且通过网络传输到Reduce节点，对其压缩可以提高很多性能，这些工作只要设置两个属性即可，我们来看下代码怎么设置。
1、给大家提供的Hadoop源码支持的压缩格式有：BZip2Codec、DefaultCodec
WordCountDriver稍微改变下，如下：
2、Mapper保持不变
3、Reducer保持不变
4.6.3 Reduce输出端采用压缩
基于WordCount案例处理。
1、修改驱动
2、Mapper和Reducer保持不变（同4.6.2）
第5章 Yarn资源调度器
Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而MapReduce等运算程序则相当于运行于操作系统之上的应用程序。
5.1 Yarn基本架构
YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成，如下图所示。
5.3 Yarn工作机制
1、Yarn运行机制，如下图所示。
2、工作机制详解：
（0）MR程序提交到客户端所在的节点。
（1）YarnRunner向ResourceManager申请一个Application。
（2）RM将该应用程序的资源路径返回给YarnRunner。
（3）该程序将运行所需资源提交到HDFS上。
（4）程序资源提交完毕后，申请运行MrAppMaster。
（5）RM将用户的请求初始化成一个Task。
（6）其中一个NodeManager领取到Task任务。
（7）该NodeManager创建容器Container，并产生MrAppmaster。
（8）Container从HDFS上拷贝资源到本地。
（9）MrAppmaster向RM申请运行MapTask资源。
（10）RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（11）MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。
（12）MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。
（13）ReduceTask向MapTask获取相应分区的数据。
（14）程序运行完毕后，MR会向RM申请注销自己。
5.4 作业提交全过程
1、作业提交过程之YARN，如下图所示。
作业提交全过程详解：
（1）作业提交
第1步：Client调用job.waitForCompletion()方法，向整个集群提交MapReduce作业。
第2步：Client向RM申请一个作业id。
第3步：RM给Client返回该job资源的提交路径和作业id。
第4步：Client提交jar包、切片信息和配置文件到指定的资源提交路径。
第5步：Client提交完资源后，向RM申请运行MrAppMaster。
（2）作业初始化
第6步：当RM收到Client的请求后，将该job添加到容量调度器中。
第7步：某一个空闲的NM领取到该Job。
第8步：该NM创建Container，并产生MrAppmaster。
第9步：下载Client提交的资源到本地。
（3）任务分配
第10步：MrAppMaster向RM申请运行多个MapTask任务资源。
第11步：RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（4）任务运行
第12步：MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。
第13步：MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。
第14步：ReduceTask向MapTask获取相应分区的数据。
第15步：程序运行完毕后，MR会向RM申请注销自己。
（5）进度和状态更新
YARN中的任务将其进度和状态(包括counter)返回给应用管理器，客户端每秒(通过mapreduce.client.progressmonitor.pollinterval设置)向应用管理器请求进度更新, 展示给用户。
（6）作业完成
除了向应用管理器请求作业进度外，客户端每5秒都会通过调用waitForCompletion()来检查作业是否完成。时间间隔可以通过mapreduce.client.completion.pollinterval来设置。作业完成之后, 应用管理器和Container会清理工作状态。作业的信息会被作业历史服务器存储以备之后用户核查。
（7）MapReduce过程
MapTask阶段：Read-Map-Collect-Spill-[Combine]-Merge
ReduceTask阶段：Copy-Merge-Sort-Reduce
2、作业提交过程之MapReduce，如下图所示。
5.5 资源调度器
目前，Hadoop作业调度器主要有三种：FIFO、Capacity Scheduler和Fair Scheduler。Hadoop2.7.2默认的资源调度器是Capacity Scheduler。
具体设置详见：yarn-default.xml文件
<property>
<description>The class to use as the resource scheduler.</description>
<name>yarn.resourcemanager.scheduler.class</name>
<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>
1、先进先出调度器（FIFO），如下图所示：
2、容量调度器（Capacity Scheduler），如下图所示：
3、公平调度器（Fair Scheduler），如下图所示：
5.6 任务的推测执行（秘籍）
1、作业完成时间取决于最慢的任务完成时间
一个作业由若干个Map任务和Reduce任务构成。因硬件老化、软件Bug等，某些任务可能运行非常慢。
思考：系统中有99%的Map任务都完成了，只有少数几个Map老是进度很慢，完不成，怎么办？
2、推测执行机制
发现拖后腿的任务，比如某个任务运行速度远慢于任务平均速度。为拖后腿任务启动一个备份任务，同时运行。谁先运行完，则采用谁的结果。
3、执行推测任务的前提条件
（1）每个Task只能有一个备份任务
（2）当前Job已完成的Task必须不小于0.05（大于5%）
（3）开启推测执行参数设置。mapred-site.xml文件中默认是打开的。
<property>
<name>mapreduce.map.speculative</name>
<value>true</value>
<description>If true, then multiple instances of some map tasks may be executed in parallel.</description>
</property>
<property>
<name>mapreduce.reduce.speculative</name>
<value>true</value>
<description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>
</property>
4、不能启用推测执行机制情况
（1）任务间存在严重的负载倾斜。
（2）特殊任务，比如任务向数据库中写数据。
5、算法原理，如下图所示