|
我们知道在执行map任务的时候,会将key/value写入内存或者磁盘。
这个时候我们在往内存写数据的时候，会根据key创建分区。
问题一：为什要创建分区?
我们如果文件很大，我们只使用一个reducer，这个reducer就要负责去所有map端取数据。那么势必会带来性能问题，而且服务器资源也没有合理利用起来
如果要合理利用，则需要多起几个reducer,那这几个reducer去map端拉取整个文件，这样的话就有这样一个问题：相同的key可能分布在不同map机器或者map文件中，每一个reducer计算出来的结果可能有问题。
现在我们对map的数据进行分区，然后我们就能保证相同的key都在一个分区上，然后reducer拉取数据的时候，所计算的结果是没有问题的。这样既保证的mapreduce执行的效率，又保证了数据的正确性。
简单一句话：分区就是更好的将map任务的结果均匀分配给reducer
问题二：怎么创建?
分区的默认实现HashPartitioner，它根据key的hashcode和Interger.
MAX_VALUE参与与运算然后 %指定的reducer数量。如果Reducer没有指定，那么默认就是一个，所以分区数也是一个，因为任何数%1=0，所以只有一个分区。
这种情况情况有时候可能某些分区数据很多，有的分区数据很少，有数据倾斜的问题，所以我们有时候需要自己定义Partitioner类
问题三 Reducer是如何知道取哪一台机器的哪一个分区的数据呢？
我们都说，一个Reducer对应着一个Map，那Reducer是如何知道从从哪一台机器的map输出文件中去拉取对应的partition的数据呢
3.1如何知道从哪一台机器去取数据？
Map任务结束就会通知MRAppMaster报告Map任务运行的状态或者输出的信息，然后Reducer启动之后，EventFetcher线程就会不断
向MRAppMaster查询这些状态或者信息，并把主机和对应的URL信息封装在MapHost里。然后Fetcher线程就知道从哪一台机器通过什么URL去获取数据
3.2如何知道取哪一个分区的数据呢？
我们知道Reducer的EventFetcher线程会向MRAppMaster查询Map任务的状态信息，然后封装在一个MapHost中。然后Fetcher线程就会从根据主机名和URL去copy数据，在这个URL中，URL格式如下：
url=13562/mapOutput?job=job_1488251187492_0003&reduce=0&map=attempt_1488251187492_0003_m_000001_0
户提供一个reduce参数，这个参数的值就是和各个map节点任务的分区对应着的，所以知道从哪个分区去获取数据
问题四 自定义分区
publicstatic
class CountryPartitionerextends
Partitioner<Text, Text> {
@Override
public intgetPartition(Text
key, Text value,
intnumPartitions) {
if (key.toString().equals("China"))
{
return
1;
}else
if (key.toString().equals("America")) {
return
2;
}
return
0;
}
}
publicstatic
class AggregateProvinceMapperextendsMapper<Text, Text, Text, Text>{
private Text
country = new
Text();
private Text
state = new
Text();
@Override
protected voidmap(Text
key, Textvalue, Mapper<Text, Text, Text, Text>.Contextcontext)
throws IOException, InterruptedException {
if (value ==null) {
return;
}
StringTokenizertokenizer =
newStringTokenizer(value.toString());
boolean
first = Boolean.TRUE;
while (tokenizer.hasMoreTokens()) {
Stringtext =
tokenizer.nextToken();
if (first) {
country.set(text);
first = Boolean.FALSE;
}else {
state.set(text);
}
}
context.write(country,state);
}
}
publicstatic
class AggregateProvinceReducerextends Reducer<Text, Text, Text, Text> {
@Override
protected voidreduce(Text
key,Iterable<Text>
values, Reducer<Text, Text, Text, Text>.Contextcontext)
throws IOException, InterruptedException {
TextformatState =
null;
for (Text
val :values) {
formatState =
newText();
formatState.set("=>"+val.toString());
context.write(key,formatState);
}
}
}
publicint
run(String[]args)
throws Exception {
Configurationconf =
newConfiguration();
Jobjob = Job.getInstance(conf,this.getClass().getSimpleName());
job.setJarByClass(AggregateProvinceInfo.class);
Pathin =
newPath(args[0]);
FileInputFormat.addInputPath(job,in);
Pathout =
newPath(args[1]);
FileOutputFormat.setOutputPath(job,out);
job.setMapperClass(AggregateProvinceMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setPartitionerClass(CountryPartitioner.class);
job.setNumReduceTasks(3);
job.setReducerClass(AggregateProvinceReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
boolean isCompleted =job.waitForCompletion(Boolean.TRUE);
return isCompleted ?0 :
1;
}
publicstatic
voidmain(String[]
args)throws Exception {
int num =new
Random().nextInt(1000);
if (args ==null ||
args.length ==
0) {
args =
new String[]{
"hdfs://hdfs-cluster/user/hadoop/input/country",
"hdfs://hdfs-cluster/user/hadoop/output/country"+num
};
}
int status =new
AggregateProvinceInfo().run(args);
System.exit(status);
} | 
转播
