大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL)

本文主要是介绍大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

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;}/*** 序列化方法*/@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}/*** 反序列化方法，注意反序列化的顺序和序列化的顺序完全一致*/@Overridepublic void readFields(DataInput in) throws IOException {upFlow = in.readLong();downFlow = in.readLong();sumFlow = in.readLong();}/*** 比较方法*/@Overridepublic 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;}@Overridepublic String toString() {return upFlow + "\t" + downFlow + "\t" + 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();@Overrideprotected 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("\t");// 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> {@Overrideprotected 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> {@Overridepublic 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();@Overrideprotected 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();@Overrideprotected 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;

这篇关于大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---