3.3.2Partition分区

要求将统计结果按照条件输出到不同文件中#xff08;分区#xff09;。

比如#xff1a;将统计结果按照手机归属地不同省份输出到不同文件中#…3.3.1Shuffle机制

Map方法之后Reduce方法之前的数据处理过程称之为Shuffle。

3.3.2Partition分区

要求将统计结果按照条件输出到不同文件中分区。

比如将统计结果按照手机归属地不同省份输出到不同文件中分区

2、默认Partitioner分区

numReduceTasks;默认分区是根据key的hashCode对ReduceTasks个数取模得到的。

用户没法控制哪个key存储到哪个分区。

3、自定义Partitioner步骤

1自定义类继承Partitioner重写getPartition()方法

public

job.setPartitionerClass(ProvincePartitioner.class);3自定义Parttition后要根据自定义Partitioner的逻辑设置相应数量的ReduceTask

task

job.setNumReduceTasks(5);4、分区总结

1如果ReduceTask的数量getPartition的结果数则会多昌盛几个空的输出文件part-r000xx

2如果1ReduceTask的数量getPartition的结果数则有一部分分区数据无处安放会Exception

3如果ReduceTask的数量1则不管MapTask端输出多少个分区文件最终结果都交给这一个ReduceTask最终也就只会产生一个结果文件part-r-00000

5、案例分析

1job.setNumReduceTasks(1);会正常运行只产生一个输出文件

会报错

手机号136、137、138、139开头都分别放到一个独立的4个文件中其他开头的放到一个文件中。

2、需求分析

org.apache.hadoop.mapreduce.Partitioner;public

class

}4、在驱动函数中增加自定义数据分区设置和ReduceTask设置

指定自定义数据分区

job.setPartitionerClass(ProvincePartitioner.class);//

task

job.setNumReduceTasks(5);package

java.io.IOException;

org.apache.hadoop.conf.Configuration;

import

org.apache.hadoop.fs.FileSystem;

import

org.apache.hadoop.mapreduce.Job;

import

org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;

import

org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import

org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public

class

指定本程序的jar包所在的本地路径job.setJarByClass(FlowsumDriver.class);//

指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(FlowCountMapper.class);job.setReducerClass(FlowCountReducer.class);//

指定mapper输出数据的kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//

指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//

input

{fs.delete(output,true);}FileInputFormat.setInputPaths(job,

input);FileOutputFormat.setOutputPath(job,

output);//

指定自定义数据分区job.setPartitionerClass(ProvincePartitioner.class);//

taskjob.setNumReduceTasks(5);//

将job中配置的相关参数以及job所用的java类所在的jar包

提交给yarn去运行boolean

job.waitForCompletion(true);System.exit(result

1);}

MapTask和ReduceTask均会对数据按照key进行排序。

该操作属于Hadoop的默认行为。

任何应用程序中的数据均会被排序而不管逻辑上是否需要。

对于MapTask它会将处理的结果暂时放到环形缓冲区中当环形缓冲区使用率达到一定阈值时再对缓冲区中的数据进行一次快速排序并将这些有序数据溢写到磁盘上而当数据处理完毕后它会对磁盘上所有文件进行归并排序。

对于ReduceTask它从每个MapTask上远程拷贝相应的数据文件如果文件大小超过一定的阈值则溢写到磁盘上否则存储在内存中。

如果磁盘上文件数目达到一定阈值则进行一次归并排序以生成一个更大的文件如果内存中文件大小或者数量超过一定阈值则进行一次合并后将数据溢写到磁盘上。

当所有数据拷贝完毕后ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

1部分排序

MapReduce根据输入记录的键对数据集排序。

保证输出的每个文件内部有序。

2全排序

最终输出结果只有一个文件且文件内部有序。

实现方式是只设置一个ReduceTask。

但该方法在处理大型文件时效率极低因为一台机器处理所有文件完全丧失了MapReduce所提供的并行架构。

在Reduce端对key进行分组。

应用于在接收的key为bean对象时想让一个或几个字段相同全部比较不相同的key进入到同一个reduce方法时可以采用分组排序。

4二次排序

在自定义排序过程中如果compareTo中的判断条件为两个即为二次排序。

1原理分析

bean对象做为key传输需要实现WritableComparable接口重写compareTo方法就可以实现排序。

Override

}3.3.5WritableComparable排序案例实操全排序

1、需求

org.apache.hadoop.io.WritableComparable;import

java.io.DataInput;

{out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}/***

反序列化方法

org.apache.hadoop.io.LongWritable;

import

org.apache.hadoop.mapreduce.Mapper;import

class

Long.parseLong(fields[2]);k.set(upFlow,

输出context.write(k,

org.apache.hadoop.mapreduce.Reducer;import

class

org.apache.hadoop.io.LongWritable;

import

org.apache.hadoop.mapreduce.Mapper;import

class

Long.parseLong(fields[fields.length

3]);long

Long.parseLong(fields[fields.length

2]);k.set(phoneNum);

}3.3.6WritableComparable排序案例实操区内排序

1需求

org.apache.hadoop.mapreduce.Partitioner;public

class

job.setPartitionerClass(ProvincePartitioner.class);//

设置Reducetask个数

job.setNumReduceTasks(5);3.3.7Combiner合并

1概述

1Combiner是MR程序中Mapper和Reducer之外的一个组件。

4Combiner的意义就是对每一个MapTask的输出进行局部汇总以减少网络传输量。

5Combiner能够应用的前提是不能影响最终的业务逻辑而且Combiner的输出kv应该跟Reducer的输入kv类型保持一致。

Mapper

a自定义一个Combiner继承Reducer重写Reduce方法

public

Text,IntWritable{Overrideprotected

void

job.setCombinerClass(WordcountCombiner.class);3.3.8Combiner合并案例实操

1、需求

统计过程中对每一个MapTask的输出进行局部汇总以减小网络传输量即采用Combiner功能。

1数据输入

1增加一个WordcountCombiner类继承Reducer

package

org.apache.hadoop.io.IntWritable;

import

org.apache.hadoop.mapreduce.Reducer;import

class

IntWritable();Overrideprotected

void

}2在WordcountDriver驱动类中指定Combiner

指定需要使用combiner以及用哪个类作为combiner的逻辑

job.setCombinerClass(WordcountCombiner.class);4案例实操-方案二

1将WordcountReducer作为Combiner在WordcountDriver驱动类中指定

指定需要使用Combiner以及用哪个类作为Combiner的逻辑

job.setCombinerClass(WordcountReducer.class);3.3.9GroupingComparator分组辅助排序

1概述

}3.3.10GroupingComparator分组案例实操

1、需求

1利用“订单id和成交金额”作为key可以将Map阶段读取到的所有订单数据按照id升序排序如果id相同再按照金额降序排序发送到Reduce。

2在Reduce端利用groupingComparator将订单id相同的kv聚合成组然后取第一个即是该订单中最贵商品如图4-18所示。

3、代码实现

org.apache.hadoop.io.WritableComparable;import

java.io.DataInput;

{out.writeInt(order_id);out.writeDouble(price);}Overridepublic

void

in.readDouble();}Overridepublic

String

org.apache.hadoop.io.LongWritable;

import

org.apache.hadoop.io.NullWritable;

import

org.apache.hadoop.mapreduce.Mapper;import

class

封装对象k.setOrder_id(Integer.parseInt(fields[0]));k.setPrice(Double.parseDouble(fields[2]));//

写出context.write(k,

}3编写OrderSortGroupingComparator类

package

org.apache.hadoop.io.WritableComparable;

import

org.apache.hadoop.io.WritableComparator;public

class

org.apache.hadoop.io.NullWritable;

import

org.apache.hadoop.mapreduce.Reducer;import

class

org.apache.hadoop.conf.Configuration;

import

org.apache.hadoop.fs.FileSystem;

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;import

class

设置jar包加载路径job.setJarByClass(OrderDriver.class);//

加载map/reduce类job.setMapperClass(OrderMapper.class);job.setReducerClass(OrderReducer.class);//

设置map输出数据key和value类型job.setMapOutputKeyClass(OrderBean.class);job.setMapOutputValueClass(NullWritable.class);//

设置最终输出数据的key和value类型job.setOutputKeyClass(OrderBean.class);job.setOutputValueClass(NullWritable.class);//

设置输入数据和输出数据路径Path

{fs.delete(output,true);}FileInputFormat.setInputPaths(job,

input);FileOutputFormat.setOutputPath(job,

output);//

设置reduce端的分组job.setGroupingComparatorClass(OrderGroupingComparator.class);//

提交boolean

job.waitForCompletion(true);System.exit(result

1);}