前言
本文主要介绍 MapReduce 的原理及开发，讲解如何利用 Combine、Partitioner、WritableComparator等组件对数据进行排序筛选聚合分组的功能。
由于文章是针对开发人员所编写的，在阅读本文前，文章假设读者已经对Hadoop的工作原理、安装过程有一定的了解，因此对Hadoop的安装就不多作说明。请确保源代码运行在Hadoop 2.x以上版本，并以伪分布形式安装以方便进行调试（单机版会对 Partitioner 功能进行限制）。
文章主要利用例子介绍如何利用 MapReduce 模仿 SQL 关系数据库进行SELECT、WHERE、GROUP、JOIN 等操作，并对 GroupingComparator、SortComparator 等功能进行说明。
希望本篇文章能对各位的学习研究有所帮助，当中有所错漏的地方敬请点评。
目录
五、WritableComparatable 自定义键值说明
一、MapReduce 工作原理简介
对Hadoop有兴趣的朋友相信对Hadoop的主要工作原理已经有一定的认识，在讲解MapReduce的程序开发前，本文先针对Mapper、Reducer、Partitioner、Combiner、Suhffle、Sort的工作原理作简单的介绍，以帮助各位更好地了解后面的内容。
图 1.1
1.1 Mapper 阶段
当系统对数据进行分片后，每个输入分片会分配到一个Mapper任务来处理，默认情况下系统会以HDFS的一个块大小64M作为分片大小，当然也可以通过配置文件设置块的大小。随后Mapper节点输出的数据将保存到一个缓冲区中（缓冲区的大小默认为512M，可通过mapreduce.task.io.sort.mb属性进行修改），缓冲区越大排序效率越高。当该缓冲区快要溢出时（缓冲区默认大小为80%，可通过mapreduce.map.sort.spill.percent属性进行修改），系统会启动一个后台线程，将数据传输到会到本地的一个文件当中。
1.2 Partitioner 阶段
在Mapper完成 KEY/VALUE 格式的数据操作后，Partitioner 将会被调用，由于真实环境中 Hadoop 可能会包含几十个甚至上百个Reducer ，Partitioner 的主要作用就是根据自定义方式确定数据将被传输到哪一个Reducer进行处理。
1.3 Combiner 阶段
如果系统定义了Combiner，在经过 Partitioner 排序处理后将会进行 Combiner处理。我们可以把 Combiner 看作为一个小型的 Reducer ，由于数据从 Mapper 通过网络传送到 Reducer ，资源开销很大，Combiner 目的就是在数据传送到Reducer前作出初步聚集处理，减少服务器的压力。如果数据量太大，还可以把 mapred.compress.map.out 设置为 true，就可以将数据进行压缩。（关于数据压缩的内容已经超越本文的讨论范围，以后会有独立的篇章针对数据压缩进行专题讨论，敬请期待）
1.4 Shuffle 阶段
在 Shuffle 阶段，每个 Reducer 会启动 5 个线程（可通过 mapreduce.reduce.shuffle.parallelcopies 进行设置）通过HTTP协议获取Mapper传送过来的数据。每次数据发送到 Reducer 前，都会根据键先进行排序。开发人员也可通过自定义的 SortComparator 进行数据排序，也是根据 GroupComparator 按照数据的其他特性进行分组处理，下面章节将会详细举例介绍。对数据进行混洗、排序完成后，将传送到对应的Reducer进行处理。
1.5 Reducer 阶段
当 Mapper 实例完成输入的数据超过设定值后（可通过mapreduce.job.reduce.slowstart.completedmaps 进行设置）, Reducer 就会开始执行。Reducer 会接收到不同 Mapper 任务传来已经过排序的数据，并通过Iterable 接口进行处理。在 Partitioner 阶段，系统已定义哪些数据将由个 Reducer 进行管理。当 Reducer 检测到 KEY 时发生变化时，系统就会按照已定的规则生成一个新的 Reducer 对数据进行处理。
如果 Reducer 端接受的数据量较小，数据则可直接存储在内存缓冲区中，方便后面的数据输出（缓冲区大小可通过mapred.job.shuffle.input.buffer.percent 进行设置）
如果数据量超过了该缓冲区大小的一定比例（可以通过 mapred.job.shuffle.merge.percent 进行设置），数据将会被合并后写到磁盘中。
二、MapReduce 开发实例
上一章节讲解了 MapReduce 的主要流程，下面将以几个简单的例子模仿 SQL 关系数据库向大家介绍一下 MapReduce 的开发过程。
HDFS常用命令 （此处只介绍几个常用命令，详细内容可在网上查找）
- 创建目录 hdfs dfs -mkdir -p 【Path】
- 复制文件 hdfs dfs -copyFromLocal 【InputPath】【OutputPath】
- 查看目录 hdfs dfs -ls 【Path】
- 运行JAR hadoop jar 【Jar名称】 【Main类全名称】 【InputPath】 【OutputPath】
2.1 使用 SELECT 获取数据
应用场景：假设在 hdfs 文件夹 / input / 20180509 路径的 *.dat 类型文件中存放在着大量不同型号的 iPhone 手机当天在不同地区的销售记录，系统想对这些记录进行统计，计算出不同型号手机的销售总数。
计算时，在Mapper中获取每一行的信息，并把iPhone名称作为Key插入，把数据作为Value插入到Context当中。
当Reducer接收到相同Key数据后，再作统一处理。
注意 ： 当前例子当中 Mapper 的输入 Key 为 LongWritable 长类型
在此过程中要注意几点： 例子中 SaleManager 继承了 org.apache.hadoop.conf.Configured 类并实现了 org.apache.hadoop.util.Tool 接口的 public static int run(Configuration conf,Tool tool, String[] args) 方法，MapReduce的相关操作都在run里面实现。由于 Configured 已经实现了 getConf() 与setConfig() 方法，创建Job时相关的配置信息就可通过getConf()方法读入。
系统可以通过以下方法注册Mapper及Reducer处理类
Job.setMapperClass(MyMapper.class);
Job.setReducerClass(MyReducer.class);
在整个运算过程当中，数据会经过筛选与计算，所以Mapper的读入信息K1,V1与Reducer的输出信息K3,V3不一定是同一格式。
org.apache.hadoop.mapreduce.Mapper<K1,V1,K2,V2>
org.apache.hadoop.mapreduce.Reducer<K2,V2,K3,V3>
当Mapper的输出的键值类型与Reduces输出的键值类型相同时，系统可以通过下面方法设置转出数据的格式
Job.setOutputKeyClass(K);
Job.setOutputValueClass(V);
当Mapper的输出的键值类型与Reduces输出的键值类型不相同时，系统则需要通过下面方法设置Mapper转出格式
Job.setMapOutputKeyClass(K);
Job.setMapOutputValueClass(V);
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString();
this.count=Integer.valueOf(data[1].toString());
this.area=data[2].toString();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
//以iPhone型号作为Key,数量为作Value传入
context.write(new Text(iPhone.getType()), new IntWritable(iPhone.getCount()));
}
}
public static class MyReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
public void reduce(Text key,Iterable<IntWritable> values,Context context)
throws IOException,InterruptedException{
int sum=0;
//对同一型号的iPhone数量进行统计
for(IntWritable val : values){
sum+=val.get();
}
context.write(key, new IntWritable(sum));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//默认情况下Reducer数量为1个（可忽略）
job.setNumReduceTasks(1);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果
2.2 使用 WHERE 对数据进行筛选
在计算过程中，并非所有的数据都适用于Reduce的计算，由于海量数据是通过网络传输的，所消耗的 I/O 资源巨大，所以可以尝试在Mapper过程中提前对数据进行筛选。以上面的数据为例，当前系统只需要计算输入参数地区的销售数据。此时只需要修改一下Mapper类，重写setup方法，通过Configuration类的 public String[] Configuration.getStrings(参数名,默认值) 方法获取命令输入的参数，再对数据进行筛选。
public static class MyMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
private String area;
@Override
public void setup(Context context){
this.area=context.getConfiguration().getStrings("area", "BeiJing")[0];
}
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
if(this.area.equals(iPhone.area))
context.write(new Text(iPhone.getType()), new IntWritable(iPhone.getCount()));
}
}
执行命令 hadoop jar 【Jar名称】 【Main类全名称】-D 【参数名=参数值】 【InputPath】 【OutputPath】
例如：hadoop jar hadoopTest-0.2.jar sun.hadoopTest.SaleManager -D area=BeiJing /tmp/input/050901 /tmp/output/050901
此时系统将选择 area 等于BeiJing 的数据进行统计
计算结果
三、利用 Partitioner 控制键值分配
3.1 深入分析 Partitioner
Partitioner 类在 org.apache.hadoop.mapreduce.Partitioner 中，通过 Job.setPartitionerClass(Class<? extends Partitioner> cls) 方法可绑定自定义的 Partitioner。若用户没有实现自定义Partitioner 时，系统将自动绑定 Hadoop 的默认类 org.apache.hadoop.mapreduce.lib.partiton.HashPartitioner 。Partitioner 包含一个主要方法是 int getPartition(K key,V value,int numReduceTasks) ，功能是控制将哪些键分配到哪个 Reducer。此方法的返回值是 Reducer 的索引值，若系统定义了4个Reducer,其返回值为0~3。numReduceTasks 侧是当前系统的 Reducer 数量，此数量可通过Job.setNumReduceTasks(int tasks) 进行设置，在伪分布环境下，其默认值为1。
注意：
在单机环境下，系统只会使用一个 Reducer，这将导致 Partitioner 缺乏意义，这也是在本文引言中强调要使用伪分布环境进行调试的原因 。
通过反编译查看 HashPartitioner ，可见系统是通过（key.hashCode() & Interger.MAX_VALUE )%numReduceTasks 方法，根据 KEY 的 HashCode 对 Reducer 数量求余方式，确定数据分配到哪一个 Reducer 进行处理的。但如果想根据用户自定义的逻辑把数据分配到对应 Reducer，单依靠 HashPartitioner 是无法实现的，此时侧需要自定义 Partitioner 。
public class HashPartitioner<K, V> extends Partitioner<K, V> {
public int getPartition(K key, V value, int numReduceTasks) {
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
}
3.2 自定义 Partitioner
在例子当中，假设系统需要把北、上、广、深4个不同的地区的iPhone销售情况分别交付给不同 Reducer 进行统计处理。我们可以自定义一个 MyPartitioner， 通过 Job.setPartitionerClass( MyPartitioner.class ) 进行绑定。通过 Job.setNumReduceTasks(4) 设置4个Reducer 。以手机类型作为KEY，把销售数据与地区作为VALUE。在 int getPartition(K key,V value,int numReduceTasks) 方法中，根据 VALUE 值的不同返回不同的索引值。
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString();
this.count=Integer.valueOf(data[1].toString());
this.area=data[2].toString();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class MyPatitional extends Partitioner<Text,Text> {
@Override
public int getPartition(Text arg0, Text arg1, int arg2) {
// TODO 自动生成的方法存根
String area=arg1.toString().split(",")[0];
// 根据不同的地区返回不同的索引值
if(area.contentEquals("BeiJing"))
return 0;
if(area.contentEquals("GuangZhou"))
return 1;
if(area.contentEquals("ShenZhen"))
return 2;
if(area.contentEquals("ShangHai"))
return 3;
return 0;
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,Text,Text>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
context.write(new Text(iPhone.getType()), new Text(iPhone.getArea()+","+iPhone.getCount().toString()));
}
}
public static class MyReducer extends Reducer<Text,Text,Text,IntWritable>{
public void reduce(Text key,Iterable<Text> values,Context context)
throws IOException,InterruptedException{
int sum=0;
//对同一型号的iPhone数量进行统计
for(Text value : values){
String count=value.toString().split(",")[1];
sum+=Integer.valueOf(count).intValue();
}
context.write(key, new IntWritable(sum));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//设置Reduce数量为4个，伪分布式情况下不设置时默认为1
job.setNumReduceTasks(4);
//注册自定义Partitional类
job.setPartitionerClass(MyPatitional.class);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果
四、利用 Combiner 提高系统性能
在前面几节所描述的例子当中，我们都是把所有的数据完整发送到 Reducer 中再作统计。试想一下，在真实环境当中，iPhone 的销售记录数以千万计，如此巨大的数据需要在 Mapper/Reducer 当中进行传输，将会耗费多少的网络资源。这么多年来 iPhone 出品的机型不过十多个，系统能否先针对同类的机型在Mapper端作出初步的聚合计算，再把计算结果发送到 Reducer。如此一来，传到 Reducer 端的数据量将会大大减少，只要在适当的情形下使用将有利于系统的性能提升。
针对此类问题，Combiner 应运而生，我们可以把 Combiner 看作为一个小型的 Reducer ，它的目的就是在数据传送到Reducer前在Mapper中作出初步聚集处理，减少服务器之间的 I/O 数据传输压力。Combiner 也继承于Reducer，通过Job.setCombinerClass(Class<? extends Reducer> cls) 方法进行注册。
下面继续以第3节的例子作为参考，系统想要在同一个Reducer中计算所有地区不同型号手机的销售情况。我们可以把地区名作为KEY，把销售数量和手机类型转换成 MapWritable 作为 VALUE。当数据输入后，不是直接把数据传输到 Reducer ，而是通过Combiner 把Mapper中不同的型号手机的销售数量进行聚合计算，把5种型号手机的销售总数算好后传输给Reducer。在Reducer中再把来源于不同 Combiner 的数据进行求和，得出最后结果。
注意 ：
MapWritable 是 系统自带的 Writable 集合类中的其中一个，它实现了 java.util.Map<Writable,Writable> 接口，以单字节充当类型数据的索引，常用于枚举集合的元素。
public class SaleManager extends Configured implements Tool{
private static IntWritable TYPE=new IntWritable(0);
private static IntWritable VALUE=new IntWritable(1);
private static IntWritable IPHONE7=new IntWritable(2);
private static IntWritable IPHONE7_PLUS=new IntWritable(3);
private static IntWritable IPHONE8=new IntWritable(4);
private static IntWritable IPHONE8_PLUS=new IntWritable(5);
private static IntWritable IPHONEX=new IntWritable(6);
public static class MyMapper extends Mapper<LongWritable,Text,Text,MapWritable>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
context.write(new Text(iPhone.getArea()), getMapWritable(iPhone.getType(), iPhone.getCount()));
}
private MapWritable getMapWritable(String type,Integer count){
Text _type=new Text(type);
Text _count=new Text(count.toString());
MapWritable mapWritable=new MapWritable();
mapWritable.put(TYPE,_type);
mapWritable.put(VALUE,_count);
return mapWritable;
}
}
public static class MyCombiner extends Reducer<Text,MapWritable,Text,MapWritable> {
public void reduce(Text key,Iterable<MapWritable> values,Context context)
throws IOException, InterruptedException{
int iPhone7=0;
int iPhone7_PLUS=0;
int iPhone8=0;
int iPhone8_PLUS=0;
int iPhoneX=0;
//对同一个Mapper所处理的不同型号的手机数据进行初步统计
for(MapWritable value:values){
String type=value.get(TYPE).toString();
Integer count=Integer.valueOf(value.get(VALUE).toString());
if(type.contentEquals("iPhone7"))
iPhone7+=count;
if(type.contentEquals("iPhone7_PLUS"))
iPhone7_PLUS+=count;
if(type.contentEquals("iPhone8"))
iPhone8+=count;
if(type.contentEquals("iPhone8_PLUS"))
iPhone8_PLUS+=count;
if(type.contentEquals("iPhoneX"))
iPhoneX+=count;
}
MapWritable mapWritable=new MapWritable();
mapWritable.put(IPHONE7, new IntWritable(iPhone7));
mapWritable.put(IPHONE7_PLUS, new IntWritable(iPhone7_PLUS));
mapWritable.put(IPHONE8, new IntWritable(iPhone8));
mapWritable.put(IPHONE8_PLUS, new IntWritable(iPhone8_PLUS));
mapWritable.put(IPHONEX, new IntWritable(iPhoneX));
context.write(key,mapWritable);
}
}
public static class MyReducer extends Reducer<Text,MapWritable,Text,Text>{
public void reduce(Text key,Iterable<MapWritable> values,Context context)
throws IOException,InterruptedException{
int iPhone7=0;
int iPhone7_PLUS=0;
int iPhone8=0;
int iPhone8_PLUS=0;
int iPhoneX=0;
//对同一地区不同型的iPhone数量进行统计
for(MapWritable value : values){
iPhone7+=Integer.parseInt(value.get(IPHONE7).toString());
iPhone7_PLUS+=Integer.parseInt(value.get(IPHONE7_PLUS).toString());
iPhone8+=Integer.parseInt(value.get(IPHONE8).toString());
iPhone8_PLUS+=Integer.parseInt(value.get(IPHONE8_PLUS).toString());
iPhoneX+=Integer.parseInt(value.get(IPHONEX).toString());
}
StringBuffer data=new StringBuffer();
data.append("iPhone7:"+iPhone7+" ");
data.append("iPhone7_PLUS:"+iPhone7_PLUS+" ");
data.append("iPhone8:"+iPhone8+" ");
data.append("iPhone8_PLUS:"+iPhone8_PLUS+" ");
data.append("iPhoneX:"+iPhoneX+" ");
context.write(key, new Text(data.toString()));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(MapWritable.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//注册Combiner处理类
job.setCombinerClass(MyCombiner.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//伪分布式情况下不设置时默认为1
job.setNumReduceTasks(1);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果
五、WritableComparable自定义键值说明
5.1 Writable、Comparable、WritableComparable 之间关系
在 Mapper 与 Reducer 中使用到的键类型、值类型都必须实现 Writable 接口，而键类型侧需要实现 WritableComparable，它们之间的关系如下图：
Writable 接口有两个方法
- write(java.io.DataOutput out) 将实例的原始属性写到 dataOutput 输出流中，其作用是序列化基础数据
- readFields（java.io.DataInput in) 从 dataInput 对象中抓取数据并重新创建 Writable
Comparable 接口中 int compareTo（object) 方法侧定义了排序的方式，如果返回值为0（判断为两个对象相等），侧被同一个 reduce 方法处理，一旦两个对象不相等，系统就会生成另一个 reduce 处理。
5.2 自定义值类型
以第三节的例子作为讨论，假设系统需要把北、上、广、深4个不同的地区的iPhone销售情况分别交付给4个不同 Reducer 节点进行统计处理。使用地区 area 作为Key，使用继承 Writable 接口的PhoneValue作为 Value 值类型，实现 write 方法与 readFields 方法，最后在 reduce 方法区分不同的型号进行计算。
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString().trim();
this.count=Integer.valueOf(data[1].toString().trim());
this.area=data[2].toString().trim();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class PhoneValue implements Writable {
public Text type=new Text();
public IntWritable count=new IntWritable();
public Text area=new Text();
public PhoneValue(){
}
public PhoneValue(String type,Integer count,String area){
this.type=new Text(type);
this.count=new IntWritable(count);
this.area=new Text(area);
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public IntWritable getCount() {
return count;
}
public void setCount(IntWritable count) {
this.count = count;
}
public Text getArea() {
return area;
}
public void setArea(Text area) {
this.area = area;
}
@Override
public void readFields(DataInput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.readFields(arg0);
this.count.readFields(arg0);
this.area.readFields(arg0);
}
@Override
public void write(DataOutput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.write(arg0);
this.count.write(arg0);
this.area.write(arg0);
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,Text,PhoneValue>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
PhoneValue phoneValue=new PhoneValue(iPhone.getType(),iPhone.getCount(),iPhone.getArea());
context.write(new Text(iPhone.getArea()), phoneValue);
}
}
public static class MyReducer extends Reducer<Text,PhoneValue,Text,Text>{
Integer iPhone7=new Integer(0);
Integer iPhone7_PLUS=new Integer(0);
Integer iPhone8=new Integer(0);
Integer iPhone8_PLUS=new Integer(0);
Integer iPhoneX=new Integer(0);
public void reduce(Text key,Iterable<PhoneValue> values,Context context)
throws IOException,InterruptedException{
//对不同类型iPhone数量进行统计
for(PhoneValue phone : values){
int count=phone.getCount().get();
if(phone.type.toString().equals("iPhone7"))
iPhone7+=count;
if(phone.type.toString().equals("iPhone7_PLUS"))
iPhone7_PLUS+=count;
if(phone.type.toString().equals("iPhone8"))
iPhone8+=count;
if(phone.type.toString().equals("iPhone8_PLUS"))
iPhone8_PLUS+=count;
if(phone.type.toString().equals("iPhoneX"))
iPhoneX+=count;
}
context.write(new Text("iPhone7"), new Text(iPhone7.toString()));
context.write(new Text("iPhone7_PLUS"), new Text(iPhone7_PLUS.toString()));
context.write(new Text("iPhone8"), new Text(iPhone8.toString()));
context.write(new Text("iPhone8_PLUS"), new Text(iPhone8_PLUS.toString()));
context.write(new Text("iPhoneX"), new Text(iPhoneX.toString()));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(PhoneValue.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//伪分布式情况下不设置时默认为1
job.setNumReduceTasks(4);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果与第三节相同
5.3 自定义键类型
Hadoop 常用的类 IntWritable、LongWritable、Text、BooleanWritable 等都实现了WritableComparable 接口，当用户需要自定义键类型时，只需要实现WritableComparable接口即可。public boolean equals(Object o) 与 public int hashCode() 都是 Object 的方法，回顾本文的第三节可以看到 hashCode 会被系统默认的 Partitioner 即 HashPartitioner 类所使用。在使用系统默认的 HashPartitioner 类时，一旦 hashCode 相等，数据将返回到同一个Reducer 节点，因此应该按业务的需求重新定义键类型的 hashCode。同时 equals 方法应该按照 hashCode 逻辑统一修改，避免在使用 Hash 散列时出现逻辑错误。
以第三节的例子作为讨论，假设系统需要把北、上、广、深4个不同的地区的iPhone销售情况分别交付给不同 Reducer 节点进行统计处理，我们只需要定义 PhoneKey 作为键类型，当中包含地区号 area 和型号 type。在 hashCode 中以地区号 area 作为指标，在 compareTo 方法中我们以手机的类型 type 进行排序。系统就可在不同的 Reducer 节点中计算出同一地点不同类型手机的销售情况。
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString().trim();
this.count=Integer.valueOf(data[1].toString().trim());
this.area=data[2].toString().trim();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class PhoneKey implements WritableComparable<PhoneKey> {
public Text type=new Text();
public Text area=new Text();
public PhoneKey(){
}
public PhoneKey(String type,String area){
this.type=new Text(type);
this.area=new Text(area);
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public Text getArea() {
return area;
}
public void setArea(Text area) {
this.area = area;
}
@Override
public void readFields(DataInput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.readFields(arg0);
this.area.readFields(arg0);
}
@Override
public void write(DataOutput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.write(arg0);
this.area.write(arg0);
}
@Override
public int compareTo(PhoneKey o) {
// TODO 自动生成的方法存根
return this.type.compareTo(o.type);
}
@Override
public boolean equals(Object o){
if(!(o instanceof PhoneKey)){
return false;
}
PhoneKey phone=(PhoneKey) o;
return this.area.equals(phone.area);
}
@Override
public int hashCode(){
return this.area.hashCode();
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,PhoneKey,IntWritable>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
PhoneKey phoneKey=new PhoneKey(iPhone.getType(),iPhone.getArea());
context.write(phoneKey, new IntWritable(iPhone.count));
}
}
public static class MyReducer extends Reducer<PhoneKey,IntWritable,Text,Text>{
public void reduce(PhoneKey phoneKey,Iterable<IntWritable> values,Context context)
throws IOException,InterruptedException{
String type=phoneKey.getType().toString();
Integer total=new Integer(0);
//对不同类型iPhone数量进行统计
for(IntWritable count : values){
total+=count.get();
}
context.write(new Text(type),new Text(total.toString()));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(PhoneKey.class);
job.setMapOutputValueClass(IntWritable.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//伪分布式情况下不设置时默认为1
job.setNumReduceTasks(4);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果与第三节相同
在这个例子中只是为了让大家更好地了解自定义键类型的使用方法，而在真实环境中，自定义键类型，主要作为区分数据的标准。如果需要更好地平衡服务器资源，分配 Reducer 数据处理的负荷，还是要通过自定义的 Partitioner 进行管理。
六、实现数据排序与分组处理
6.1 RawComparator 接口介绍
在实际的应用场景当中，很可能会用到第三方类库作为键类型，但我们无法直接对源代码进行修改。为此系统定义了 RawComparator 接口，假设第三方类已实现了 Writable 接口，用户可通过自定义类实现 RawComparator 接口，通过 job.setSortComparatorClass(rawComparator.class) 设置即可。RawComparator 继承了 java.util.Comparator 接口，并添加了 int compare(byte[] b1,int s1, int l1,byte[] b2 ,int s2, int l2) 方法。此方法最简单的实现方式是通过 Writable 实例中的 readField 重构对象，然后使用通用类的 compareTo 完成排序。
public interface RawComparator<T> extends Comparator<T> {
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2);
}
下面例子假设 PhoneWritable 是第三方类库中的值类型，我们无法直接修改，但系统需要把 PhoneWritable 用作 KEY 处理，按照不同地区不同型号进行排序计算出手机的销售情况。此时可建立 PhoneComparator 类并实现 RawComparator 接口，在主程序中通过 job.setSortComparatorClass(PhoneComparator.class) 设置此接口的实现类。
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString().trim();
this.count=Integer.valueOf(data[1].toString().trim());
this.area=data[2].toString().trim();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class PhoneWritable implements Writable {
public Text type=new Text();
public IntWritable count=new IntWritable();
public Text area=new Text();
public PhoneWritable(){
}
public PhoneWritable(String type,Integer count,String area){
this.type=new Text(type);
this.count=new IntWritable(count);
this.area=new Text(area);
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public IntWritable getCount() {
return count;
}
public void setCount(IntWritable count) {
this.count = count;
}
public Text getArea() {
return area;
}
public void setArea(Text area) {
this.area = area;
}
@Override
public void readFields(DataInput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.readFields(arg0);
this.count.readFields(arg0);
this.area.readFields(arg0);
}
@Override
public void write(DataOutput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.write(arg0);
this.count.write(arg0);
this.area.write(arg0);
}
}
public class PhoneComparator implements RawComparator<PhoneWritable> {
private DataInputBuffer buffer=null;
private PhoneWritable phone1=null;
private PhoneWritable phone2=null;
public PhoneComparator(){
buffer=new DataInputBuffer();
phone1=new PhoneWritable();
phone2=new PhoneWritable();
}
@Override
public int compare(PhoneWritable o1, PhoneWritable o2) {
// TODO 自动生成的方法存根
if(!o1.getArea().equals(o2.getArea()))
return o1.getArea().compareTo(o2.getArea());
else
return o1.getType().compareTo(o2.getType());
}
@Override
public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3, int arg4, int arg5) {
// TODO 自动生成的方法存根
try {
buffer.reset(arg0,arg1,arg2);
phone1.readFields(buffer);
buffer.reset(arg3,arg4,arg5);
phone2.readFields(buffer);
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
return this.compare(phone1, phone2);
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,PhoneWritable,IntWritable>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
PhoneWritable phone=new PhoneWritable(iPhone.getType(),iPhone.getCount(),iPhone.getArea());
context.write(phone, phone.getCount());
}
}
public static class MyReducer extends Reducer<PhoneWritable,IntWritable,Text,Text>{
public void reduce(PhoneWritable phone,Iterable<IntWritable> values,Context context)
throws IOException,InterruptedException{
//对不同类型iPhone数量进行统计
Integer total=new Integer(0);
for(IntWritable count : values){
total+=count.get();
}
context.write(new Text(phone.getArea()+" "+phone.getType()),new Text(total.toString()));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(PhoneWritable.class);
job.setSortComparatorClass(PhoneComparator.class);
job.setMapOutputValueClass(IntWritable.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//伪分布式情况下不设置时默认为1
job.setNumReduceTasks(1);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果
6.2 WritableComparator 类介绍
WritableComparator 是系统自带的接口 RawComparartor 实现类，它实现了 RawComparator 接口的两个基础方法 int compare(object , object ) 与 int compare(byte[] b1,int s1, int l1,byte[] b2 ,int s2, int l2)
通过反编译查看源代码可知道，系统也是通过 WritableComparable 接口的 readField 方法重构对象，然后调用 int compareTo (WritableComparable,WritableComparable) 方法完成排序的。因此一般情况下我们在继承 WritableComparator 类实现排序时，只需要重构此方法实现业务逻辑即可。
6.3 利用 WritableComparator 实现数据排序
假设系统原有的键类型 PhoneKey 是以手机类型 type作为排序标准，现在我们需要通过 WritableComparator 把排序标准修改为按先按地区 area 再按类型 type 排序。按照第上节所述，我们只需要继承 WritableComparator 类，重写 int compareTo (WritableComparable,WritableComparable)，按照地区 area 及 类型 type 进行排序，最后使用 job.setSortComparatorClass(Class<? extends RawComparator> cls) 设置排序方式即可。
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString().trim();
this.count=Integer.valueOf(data[1].toString().trim());
this.area=data[2].toString().trim();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class PhoneKey implements WritableComparable<PhoneKey> {
public Text type=new Text();
public Text area=new Text();
public PhoneKey(){
}
public PhoneKey(String type,String area){
this.type=new Text(type);
this.area=new Text(area);
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public Text getArea() {
return area;
}
public void setArea(Text area) {
this.area = area;
}
@Override
public void readFields(DataInput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.readFields(arg0);
this.area.readFields(arg0);
}
@Override
public void write(DataOutput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.write(arg0);
this.area.write(arg0);
}
@Override
public int compareTo(PhoneKey o) {
// TODO 自动生成的方法存根
return this.type.compareTo(o.type);
}
@Override
public boolean equals(Object o){
if(!(o instanceof PhoneKey)){
return false;
}
PhoneKey phone=(PhoneKey) o;
return this.area.equals(phone.area);
}
@Override
public int hashCode(){
return this.area.hashCode();
}
}
public class PhoneSortComparator extends WritableComparator{
public PhoneSortComparator(){
super(PhoneKey.class,true);
}
@Override
public int compare(WritableComparable a,WritableComparable b){
PhoneKey key1=(PhoneKey) a;
PhoneKey key2=(PhoneKey) b;
if(!key1.getArea().equals(key2.getArea()))
return key1.getArea().compareTo(key2.getArea());
else
return key1.getType().compareTo(key2.getType());
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,PhoneKey,IntWritable>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
PhoneKey phone=new PhoneKey(iPhone.getType(),iPhone.getArea());
context.write(phone, new IntWritable(iPhone.getCount()));
}
}
public static class MyReducer extends Reducer<PhoneKey,IntWritable,Text,Text>{
public void reduce(PhoneKey phone,Iterable<IntWritable> values,Context context)
throws IOException,InterruptedException{
//对不同类型iPhone数量进行统计
Integer total=new Integer(0);
for(IntWritable count : values){
total+=count.get();
}
context.write(new Text(phone.getArea()+" "+phone.getType()+": "),new Text(total.toString()));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(PhoneKey.class);
job.setMapOutputValueClass(IntWritable.class);
//设置排序类型 SortComparator
job.setSortComparatorClass(PhoneSortComparator.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//伪分布式情况下不设置时默认为1
job.setNumReduceTasks(1);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
从计算结果可以看到数据是先按照地区 area 再按手机型号 type 进行排序的
6.4 利用 WritableComparator 实现数据分组
在 6.3 节的例子中，数据是先按照地区号 area 再按手机类型 type 进行排序的，因此在 reduce 方法中根据 Iterable 集合计算出来的将会同一地区同一类型的手机。若此时需要对同一地区所有手机类型的销售情况进行合计，可以使用 GroupingComparator 分组计算方式 。其方法是继承 WritableComparator 类，重写 int compareTo (WritableComparable,WritableComparable)，以地区号 area 作为分组标识，最后使用 job.setGroupComparatorClass(Class<? extends RawComparator> cls) 设置分组方式即可。
public class Phone {
public String type;
public Integer count;
public String area;
public Phone(String line){
String[] data=line.split(",");
this.type=data[0].toString().trim();
this.count=Integer.valueOf(data[1].toString().trim());
this.area=data[2].toString().trim();
}
public String getType(){
return this.type;
}
public Integer getCount(){
return this.count;
}
public String getArea(){
return this.area;
}
}
public class PhoneKey implements WritableComparable<PhoneKey> {
public Text type=new Text();
public Text area=new Text();
public PhoneKey(){
}
public PhoneKey(String type,String area){
this.type=new Text(type);
this.area=new Text(area);
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public Text getArea() {
return area;
}
public void setArea(Text area) {
this.area = area;
}
@Override
public void readFields(DataInput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.readFields(arg0);
this.area.readFields(arg0);
}
@Override
public void write(DataOutput arg0) throws IOException {
// TODO 自动生成的方法存根
this.type.write(arg0);
this.area.write(arg0);
}
@Override
public int compareTo(PhoneKey o) {
// TODO 自动生成的方法存根
return this.type.compareTo(o.type);
}
@Override
public boolean equals(Object o){
if(!(o instanceof PhoneKey)){
return false;
}
PhoneKey phone=(PhoneKey) o;
return this.area.equals(phone.area);
}
@Override
public int hashCode(){
return this.area.hashCode();
}
}
public class PhoneSortComparator extends WritableComparator{
public PhoneSortComparator(){
super(PhoneKey.class,true);
}
@Override
public int compare(WritableComparable a,WritableComparable b){
PhoneKey key1=(PhoneKey) a;
PhoneKey key2=(PhoneKey) b;
if(!key1.getArea().equals(key2.getArea()))
return key1.getArea().compareTo(key2.getArea());
else
return key1.getType().compareTo(key2.getType());
}
}
public class PhoneGroupComparator extends WritableComparator{
public PhoneGroupComparator(){
super(PhoneKey.class,true);
}
@Override
public int compare(WritableComparable a,WritableComparable b){
PhoneKey key1=(PhoneKey) a;
PhoneKey key2=(PhoneKey) b;
return key1.getArea().compareTo(key2.getArea());
}
}
public class SaleManager extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,PhoneKey,IntWritable>{
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String data=value.toString();
Phone iPhone=new Phone(data);
PhoneKey phone=new PhoneKey(iPhone.getType(),iPhone.getArea());
context.write(phone, new IntWritable(iPhone.getCount()));
}
}
public static class MyReducer extends Reducer<PhoneKey,IntWritable,Text,Text>{
public void reduce(PhoneKey phone,Iterable<IntWritable> values,Context context)
throws IOException,InterruptedException{
//对不同类型iPhone数量进行统计
Integer total=new Integer(0);
for(IntWritable count : values){
total+=count.get();
}
context.write(new Text(phone.getArea()),new Text(total.toString()));
}
}
public int run(String[] arg0) throws Exception {
Job job=Job.getInstance(getConf());
job.setJarByClass(SaleManager.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(PhoneKey.class);
job.setMapOutputValueClass(IntWritable.class);
//设置排序类型 SortComparator
job.setSortComparatorClass(PhoneSortComparator.class);
//设置分组类型GroupComparator
job.setGroupingComparatorClass(PhoneGroupComparator.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//伪分布式情况下不设置时默认为1
job.setNumReduceTasks(1);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new SaleManager(), args);
}
}
计算结果
注意：一般情况下 job.setSortComparatorClass(Class<? extends RawComparator> cls) 与 job.setGroupComparatorClass(Class<? extends RawComparator> cls) 应该同时调用。若只设置了排序方式 SortComparator 而没有调用 job.setGroupComparatorClass(Class<? extends RawComparator> cls) 方法，则 GroupComparator 分组方式视为与 SortComparator 一致。
这也是 6.3 节在没有设置 GroupComparator 的情况下系统会按照 area 与 type 进行分组计算的原因。
七、数据集连接处理方式介绍
在处理关系数据库时，经常会遇到外连接，内连接等复杂数据查询，在 Hadoop 的数据集处理上同样会遇上类似问题。当数据源来源于不同数据集时，Hadoop 框架提供了2种不同的方法实现合并查询。
- Map 端连接查询：当两个数据集中有一个非常小而另一个非常大时，我们可以利用 DistributeCache 做缓存处理，把较小的数据集加载到缓存，在 Map 加载较大的数据源时，从缓存中查找对应的扩展数据，一同发送到 Reduce 端。
- Reduce 端连接查询：当两个数据集中的数据都非常大时，在 Map 端已经无法完全加载其中一个数据集时，我们可以设置不同的 Mapper 数据读入类，把连接键作为 Mapper 的输出键。为了在 Reduce 中实现连接，注意设置 GroupingComparator 时按需要把连接键的属性作为分组处理的标识，这样就能确保两个数据集中相同连接键的数据会被同一个 reduce 方法处理。
7.1 Map 端连接介绍
假设有这样一个使用场景：在 *.gds 结尾的数据集中记录了所有手机的编号 number、类型 type 、单价 price，由于手机类型有限，所以数据量较小。在 *.sal 结尾的数据集中记录了每个客户名称 name 与其所购买的手机编号 number，数量 count，总体价格 total，由于订单数据具大，所以数据量比较庞大。此时，我们可以在 map 方法运行前，在 setup 方法中通过 job.addCacheFile（URI）把手机型号的数据加载到缓存当中，在读入销售订单数据时，从缓存中查询对应的手机型号，合并数据后一同发送到 Reduce 端。
*.gds 数据
*.sal 数据
public class Goods {
public Text number;
public Text type;
public IntWritable price;
public Goods(String[] dataline){
this.number=new Text(dataline[0]);
this.type=new Text(dataline[1]);
this.price=new IntWritable(Integer.valueOf(dataline[2]));
}
public Text getNumber() {
return number;
}
public void setNumber(Text number) {
this.number = number;
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public IntWritable getPrice() {
return price;
}
public void setPrice(IntWritable price) {
this.price = price;
}
}
public class OrderWritable implements Writable{
public Text name=new Text();
public Text number=new Text();
public Text type=new Text();
public IntWritable price=new IntWritable();
public IntWritable count=new IntWritable();
public IntWritable total=new IntWritable();
public OrderWritable(){
}
public OrderWritable(String[] data,Goods goods){
this.name=new Text(data[0]);
this.number=new Text(data[1]);
this.type=goods.getType();
this.price=goods.getPrice();
this.count=new IntWritable(new Integer(data[2]));
this.total=new IntWritable(new Integer(data[3]));
}
public Text getName() {
return name;
}
public void setName(Text name) {
this.name = name;
}
public Text getNumber() {
return number;
}
public void setNumber(Text number) {
this.number = number;
}
public Text getType() {
return type;
}
public void setType(Text type) {
this.type = type;
}
public IntWritable getPrice() {
return price;
}
public void setPrice(IntWritable price) {
this.price = price;
}
public IntWritable getCount() {
return count;
}
public void setCount(IntWritable count) {
this.count = count;
}
public IntWritable getTotal() {
return total;
}
public void setTotal(IntWritable total) {
this.total = total;
}
@Override
public void readFields(DataInput in) throws IOException {
// TODO 自动生成的方法存根
this.name.readFields(in);
this.number.readFields(in);
this.type.readFields(in);
this.price.readFields(in);
this.count.readFields(in);
this.total.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
// TODO 自动生成的方法存根
this.name.write(out);
this.number.write(out);
this.type.write(out);
this.price.write(out);
this.count.write(out);
this.total.write(out);
}
}
public class MapJoinExample extends Configured implements Tool{
public static class MyMapper extends Mapper<LongWritable,Text,Text,OrderWritable>{
private Map<String,Goods> map=new HashMap<String,Goods>();
private Configuration conf;
//读取该URI下的文件，把文件中的goods放入map中存储
private void read(URI uri){
try {
FileSystem file=FileSystem.get(uri,conf);
FSDataInputStream hdfsInStream = file.open(new Path(uri));
InputStreamReader isr = new InputStreamReader(hdfsInStream);
BufferedReader br = new BufferedReader(isr);
String line;
while ((line = br.readLine()) != null) {
Goods goods=new Goods(line.split(","));
map.put(goods.getNumber().toString(), goods);
}
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
}
public void setup(Context context){
//获取配置
conf=context.getConfiguration();
//按照输入路径读取文件
try{
URI[] uris=context.getCacheFiles();
if(uris[0].toString().endsWith("gds")){
read(uris[0]);
}
}catch(Exception ex){
throw new RuntimeException(ex);
}
}
public void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException{
String[] datalist=value.toString().split(",");
Goods goods=map.get(datalist[1]);
if(goods!=null){
OrderWritable order=new OrderWritable(datalist,goods);
context.write(goods.getNumber(), order);
}
}
}
public static class MyReducer extends Reducer<Text,OrderWritable,Text,Text>{
public void reduce(Text number,Iterable<OrderWritable> orders,Context context)
throws IOException,InterruptedException{
for(OrderWritable order : orders)
context.write(number,new Text(order.getName()+","+order.getType()
+","+order.getPrice().toString()+","+order.getCount().toString()
+","+order.getTotal().toString()));
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(MapJoinExample.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(OrderWritable.class);
//注册Mapper及Reducer处理类
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置读入文件路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置转出文件路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
//加入缓存文件
job.addCacheFile(new URI(args[2]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new MapJoinExample(), args);
}
}
执行命令 hadoop jar 【Jar名称】 【Main类全名称】【InputPath】 【OutputPath】 【*.gds 数据的URI】
当中最后一个参数正是 *.gds 数据集的 HDFS 路径，可见执行结果如下：
7.2 Reduce 端连接介绍
假设有以下的一个应用场景，在 Hadoop 数据集中存储了商品的订单数据 *.odr ，当中包含了订单号 orderCode，商品 goods，单价 price，数量 count，总体价格 total。在另一个数据集中存储了商品的送货信息 *.snd，当中包含中订单号 orderCode，商品 goods，送货地址 address，收货人 name，电话号码 tel。由于数据集的数据是一对一关系，所以数据量都非常具大，无法在 Mapper 端实现缓存扩展的方式。此时，可以通过系统提供的 MultipleInputs 类实现多个 Mapper 数据输入，不同的数据格式由不同的 Mapper 类进行处理。通过设置 GroupingComparator 按需要把连接键的属性作为分组处理的标识，最后在 Reduce 端把具有相同特性的数据进行合并处理。
*.odr 订单数据
*.snd 送货数据
我们把订单号 orderCode 与商品 goods 作为了键的两个属性，目的在排序时先按商品类型再按订单号进行排序，而 type 值是用于区分此键的数据是来源于订单数据集还是送货单数据集。在设置 GroupComparator 时，我们把商品 goods 作为标识，把同一类商品的订单交付到同一个reduce方法中进行处理。最后通过 MultipleInputs.addInputPath
(Job job, Path path, Class<? extends InputFormat> inputFormatClass, Class<? extends Mapper> mapperClass) 方法绑定不同数据集的 Mapper 处理方式。
public class DispatchKey implements WritableComparable<DispatchKey> {
public static final IntWritable TYPE_ORDER=new IntWritable(0);
public static final IntWritable TYPE_SEND=new IntWritable(1);
////数据类型，当前数据类型为订单时为0,当前数据类型为送货单时为1
public IntWritable type=new IntWritable();
public Text orderCode=new Text();
public Text goods=new Text();
@Override
public void readFields(DataInput in) throws IOException {
// TODO 自动生成的方法存根
this.type.readFields(in);
this.orderCode.readFields(in);
this.goods.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
// TODO 自动生成的方法存根
this.type.write(out);
this.orderCode.write(out);
this.goods.write(out);
}
@Override
public int compareTo(DispatchKey key) {
// 先按商品类型，再按订单号进行排序
if(this.goods.equals(key.goods)){
if(this.orderCode.equals(key.orderCode))
return this.type.compareTo(key.type);
else
return this.orderCode.compareTo(key.orderCode);
}
else
return this.goods.compareTo(key.goods);
}
@Override
public boolean equals(Object o){
if(!(o instanceof DispatchKey)){
return false;
}
DispatchKey key=(DispatchKey) o;
if(this.orderCode.equals(key.orderCode)&&this.goods.equals(key.orderCode)&&
(this.type.get()==key.type.get()))
return true;
return false;
}
@Override
public int hashCode(){
return (this.orderCode.toString()+this.goods.toString()
+this.type.toString()).hashCode();
}
}
public class DispatchValue implements Writable {
public Text order=new Text();
public IntWritable price=new IntWritable();
public IntWritable count=new IntWritable();
public IntWritable total=new IntWritable();
public Text address=new Text();
public Text name=new Text();
public Text tel=new Text();
@Override
public void readFields(DataInput in) throws IOException {
// TODO 自动生成的方法存根
this.order.readFields(in);
this.price.readFields(in);
this.count.readFields(in);
this.total.readFields(in);
this.address.readFields(in);
this.name.readFields(in);
this.tel.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
// TODO 自动生成的方法存根
this.order.write(out);
this.price.write(out);
this.count.write(out);
this.total.write(out);
this.address.write(out);
this.name.write(out);
this.tel.write(out);
}
}
public class DispatchSortComparator extends WritableComparator{
public DispatchSortComparator(){
super(DispatchKey.class,true);
}
}
public class DispatchGroupComparator extends WritableComparator{
public DispatchGroupComparator(){
super(DispatchKey.class,true);
}
@Override
public int compare(WritableComparable a,WritableComparable b){
DispatchKey key1=(DispatchKey) a;
DispatchKey key2=(DispatchKey) b;
return key1.goods.compareTo(key2.goods);
}
}
public class MapJoinExample extends Configured implements Tool{
public static class OrderMapper extends Mapper<LongWritable,Text,DispatchKey,DispatchValue>{
public void map(LongWritable longwritable,Text text,Context context)
throws IOException,InterruptedException{
String[] data=text.toString().split(",");
DispatchKey key=new DispatchKey();
//设置KEY的类型为订单
key.type=DispatchKey.TYPE_ORDER;
key.orderCode=new Text(data[0]);
key.goods=new Text(data[1]);
DispatchValue value=new DispatchValue();
value.order=new Text(data[0]);
value.price=new IntWritable(new Integer(data[2]));
value.count=new IntWritable(new Integer(data[3]));
value.total=new IntWritable(new Integer(data[4]));
context.write(key, value);
}
}
public static class SendMapper extends Mapper<LongWritable,Text,DispatchKey,DispatchValue>{
public void map(LongWritable longwritable,Text text,Context context)
throws IOException,InterruptedException{
String[] data=text.toString().split(",");
DispatchKey key=new DispatchKey();
//设置KEY类型为送货单
key.type=DispatchKey.TYPE_SEND;
key.orderCode=new Text(data[0]);
key.goods=new Text(data[1]);
DispatchValue value=new DispatchValue();
value.order=new Text(data[0]);
value.address=new Text(data[2]);
value.name=new Text(data[3]);
value.tel=new Text(data[4]);
context.write(key, value);
}
}
public static class MyReducer extends Reducer<DispatchKey,DispatchValue,Text,Text>{
public void reduce(DispatchKey key,Iterable<DispatchValue> values,Context context)
throws IOException,InterruptedException{
String mes="unknow";
String orderCode="unknow";
for(DispatchValue value : values){
//当数据为订单数据时进行记录
if(key.type.equals(DispatchKey.TYPE_ORDER)){
orderCode=key.orderCode.toString();
mes=key.goods.toString()+","+value.price.toString()+","
+value.count.toString()+","+value.total.toString()+",";
}//当数据为送货数据且订单号相等时追加记录
else if(key.type.equals(DispatchKey.TYPE_SEND)
&&key.orderCode.toString().equals(orderCode)){
mes+=value.name.toString()+","+value.address.toString()+","+value.tel.toString();
context.write(key.orderCode, new Text(mes));
//清空记录
orderCode="unknow";
mes="unkonw";
}
}
}
}
public int run(String[] arg0) throws Exception {
// TODO 自动生成的方法存根
// TODO Auto-generated method stub
Job job=Job.getInstance(getConf());
job.setJarByClass(MapJoinExample.class);
//注册Key/Value类型为Text
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//若Map的转出Key/Value不相同是需要分别注册
job.setMapOutputKeyClass(DispatchKey.class);
job.setMapOutputValueClass(DispatchValue.class);
//设置排序类型 SortComparator
job.setSortComparatorClass(DispatchSortComparator.class);
//设置分组类型
job.setGroupingComparatorClass(DispatchGroupComparator.class);
//注册Mapper及Reducer处理类
//job.setMapperClass(OrderMapper.class);
job.setReducerClass(MyReducer.class);
//输入输出数据格式化类型为TextInputFormat
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
//获取命令参数
String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
//设置Order数据处理Mapper
MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class,OrderMapper.class);
//设置Send数据处理Mapper
MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class,SendMapper.class);
//设置输出路径
FileOutputFormat.setOutputPath(job, new Path(args[2]));
boolean status=job.waitForCompletion(true);
if(status)
return 0;
else
return 1;
}
public static void main(String[] args) throws Exception{
Configuration conf=new Configuration();
ToolRunner.run(new MapJoinExample(), args);
}
}
执行命令 hadoop jar 【Jar名称】 【Main类全名称】【OrderMapperInputPath】 【SendMapperInputPath】【OutputPath】 可以到以下计算结果。
注意系统是通过 String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs() 来获取输入参数的，所以执行时注意参数的输入顺序与代码获取参数时保持一致。
本章主要介绍了 MapReduce 的开发原理及应用场景，讲解如何利用 Combine、Partitioner、WritableComparable、WritableComparator 等组件对数据进行排序筛选聚合分组的功能。对多数据集的连接查询进行分析，介绍如何通过 Map 端与 Reduce 端对多数据集连接进行处理。
后面的文章将会进一步对 Apache Hive 的应用，HBase 的集成进行讲解，敬请期待。
希望本篇文章能对各位的学习研究有所帮助，由于时间比较仓促，当中有所错漏的地方欢迎点评。
对 JAVA 开发有兴趣的朋友欢迎加入QQ群： 共同探讨！
对 .NET 开发有兴趣的朋友欢迎加入QQ群：230564952 共同探讨 ！
Hadoop 综合揭秘
MapReduce 基础编程（介绍 Combine、Partitioner、WritableComparable、WritableComparator 使用方式）
作者：风尘浪子
https://www.cnblogs.com/leslies2/p/9009574.html
原创作品，转载时请注明作者及出处