`
dajuezhao
  • 浏览: 60856 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

Map/Reduce的GroupingComparator排序简述

阅读更多

一、背景

排序对于MR来说是个核心内容,如何做好排序十分的重要,这几天写了一些,总结一下,以供以后读阅。

二、准备

1、hadoop版本是0.20.2

2、输入的数据格式(这个很重要,看清楚格式),名称是secondary.txt:
abc 123
acb 124
cbd 523
abc 234
nbc 563
fds 235
khi 234
cbd 675
fds 971
hka 862
ubd 621
khi 123
fds 321

仔细看下,数据文件第一列是字母,第二列是数字,我要做的就是结合这组数据进行一些排序的测试。

3、代码框架,因为接下来的测试改动都是针对部分代码的修改,框架的代码是不会改变的,所以先把主要代码贴在这里。

代码分为2部分:自定义的key和主框架代码(注意看下红色部分)。先贴上主框架代码:

MyGrouping.java
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.run.lenged.business.TextPair;

public class MyGrouping {

/**
* Map
*
* @author Administrator
*/
public static class MyGroupingMap extends Mapper<LongWritable, Text, TextPair, Text> {
protected void map(LongWritable key, Text value,
org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, TextPair, Text>.Context context)
throws java.io.IOException, InterruptedException {
String arr[] = value.toString().split("\t");
if (arr.length != 2) {
return;
}
TextPair tp = new TextPair();
tp.set(new Text(arr[0]), new Text(arr[1]));
context.write(tp, new Text(arr[1]));
}
}

/**
* 按照Hashcode值来进行切分
*
* @author Administrator
*/
public static class MyGroupingPartition extends Partitioner<TextPair, Text> {
@Override
public int getPartition(TextPair key, Text value, int numPartitions) {
return (key.hashCode() & Integer.MAX_VALUE) % numPartitions;
}
}

/**
* group进行排序
*
* @author Administrator
*/
@SuppressWarnings("unchecked")
public static class MyGroupingGroup extends WritableComparator {
//代码变动部分
}

/**
* reduce
*
* @author Administrator
*/
public static class MyGroupingReduce extends Reducer<TextPair, Text, Text, Text> {
protected void reduce(TextPair key, java.lang.Iterable<Text> value,
org.apache.hadoop.mapreduce.Reducer<TextPair, Text, Text, Text>.Context context)
throws java.io.IOException, InterruptedException {
StringBuffer sb = new StringBuffer();
while (value.iterator().hasNext()) {
sb.append(value.iterator().next().toString() + "_");
}
context.write(key.getFirst(), new Text(sb.toString().substring(0, sb.toString().length() - 1)));
}
}

public static void main(String args[]) throws Exception {
Configuration conf = new Configuration();
GenericOptionsParser parser = new GenericOptionsParser(conf, args);
String[] otherArgs = parser.getRemainingArgs();
if (args.length != 2) {
System.err.println("Usage: NewlyJoin <inpath> <output>");
System.exit(2);
}

Job job = new Job(conf, "MyGrouping");
// 设置运行的job
job.setJarByClass(MyGrouping.class);
// 设置Map相关内容
job.setMapperClass(MyGroupingMap.class);
job.setMapOutputKeyClass(TextPair.class);
job.setMapOutputValueClass(Text.class);
job.setPartitionerClass(MyGroupingPartition.class);

job.setGroupingComparatorClass(MyGroupingGroup.class);

// 设置reduce
job.setReducerClass(MyGroupingReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
// 设置输入和输出的目录
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
// 执行,直到结束就退出
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
TextPair.java
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

public class TextPair implements WritableComparable<TextPair> {

private Text first;
private Text second;

public TextPair() {
set(new Text(), new Text());
}

public void set(Text first, Text second) {
this.first = first;
this.second = second;
}

public Text getFirst() {
return first;
}

public Text getSecond() {
return second;
}

@Override
public void readFields(DataInput in) throws IOException {
first.readFields(in);
second.readFields(in);
}

@Override
public void write(DataOutput out) throws IOException {
first.write(out);
second.write(out);
}

@Override
public int compareTo(TextPair o) {
int cmp = first.compareTo(o.first);
if (cmp != 0) {
return cmp;
} else {
return second.compareTo(o.second);
}
}
}
三、测试前提
1、首先提一个需求,我们结合需求来测试,然后再扩散开。
需求内容是:如果第一列值相同,第二列值叠加,并对第二列值进行升序排序。最后输出的时候,按照第一列值的升序排序输出。
2、需求实现。
根据上面的需求,我们可以分析一下:
需要对第一个字段和第二个字段都进行排序,那么单纯的利用MR框架对key迭代输出,value累加是不行的。因为value是没有进行排序。
所以我们需要做一些改动,定义key为符合组建。TextPair.java类就是自定义的key。
一般来说如果要对key和value同时做排序,那么,自定义的组合key的格式第一个值是第一个字段,第二个值就是第二个字段。
3、那么我们就定义一个job.setGroupingComparatorClass(MyGroupingGroup.class);代码如下:
public static class MyGroupingGroup extends WritableComparator {
protected MyGroupingGroup() {
super(TextPair.class, true);
}

@Override
public int compare(WritableComparable a, WritableComparable b) {
TextPair mip1 = (TextPair) a;
TextPair mip2 = (TextPair) b;
return mip1.getFirst().compareTo(mip2.getFirst());
}
}

只对输出的复合组建第一项值进行排序。输出的结果如下:
abc 123_234
acb 124
cbd 523_675
fds 235_321_971
hka 862
khi 123_234
nbc 563
ubd 621

4、查看结果,我们可以看出,基本满足了上面的需求。那么接下来,我们就将做个测试,来实现一下MR的排序功能。

四、Group按第二个字段值进行排序测试

1、修改一下group的排序方式,针对第二个值进行合并排序,代码如下:
public static class MyGroupingGroup extends WritableComparator {
protected MyGroupingGroup() {
super(TextPair.class, true);
}

@Override
public int compare(WritableComparable a, WritableComparable b) {
TextPair mip1 = (TextPair) a;
TextPair mip2 = (TextPair) b;
return mip1.getSecond().compareTo(mip2.getSecond());
//return mip1.getFirst().compareTo(mip2.getFirst());
}
}

2、reduce的输出稍微改下,将第2个字段也输出,方便查看,代码如下:

context.write(key.getFirst(), new Text(sb.toString().substring(0, sb.toString().length() - 1)));

reduce输出的结果:
abc_123 123
abc_234 234
acb_124 124
cbd_523 523
cbd_675 675
fds_235 235
fds_321 321
fds_971 971
hka_862 862
khi_123 123
khi_234 234
nbc_563 563
ubd_621 621

3、看到结果,第一反应就是没有按照我的要求,按第二个值进行排序操作。

其实不是,这个结果确实是进行了group的排序,只是说遇到没有符合合并结果数据。所以,看起来没有进行排序。

在这里有个概念,就是group到底是在什么时候做的排序,原文是这样写的:

Job.setGroupingComparatorClass(Class<? extends RawComparator> cls)
Define the comparator that controls which keys are grouped together
for a single call to Reducer.reduce(Object, Iterable, org.apache.hadoop.mapreduce.Reducer.Context)

我尝试翻译了一下(英文水平实在是有限,不对的地方还望各位指出):

在一个reduce的调用过程中,定义一个comparator,对分组在一起的key进行排序。

通过上面这句话就可以理解,为什么khi_123 123和abc_123 123没有叠加在一起。

五、总结

1、这里只写了group的排序,没有写sort,后面将会写一个,说不定就是今天晚上吧!

2、过几天写个MR的执行流程,并画个图,贴出来大家看看。

3、对于这块的排序我也是接触不久,可能有写的不对的地方。还望朋友们跟贴指出来。

4、如果有疑问或是不好跟贴,可以发邮件交流:dajuezhao@gmail.com

分享到:
评论

相关推荐

    Map/Reduce

    Map/Reduce介绍。一些基本基础介绍。

    Map/Reduce:大规模集群上的简化数据处理

    该模型的核心思想是将复杂的并行和分布式计算过程简化为两个主要步骤:Map 和 Reduce。Map 阶段将原始数据拆分成键值对,应用用户自定义的映射函数,生成中间键值对;Reduce 阶段则将具有相同中间键的值聚合,通过...

    基于Map/Reduce的分布式搜索引擎研究

    ### 基于Map/Reduce的分布式搜索引擎研究 #### 1. 引言 随着互联网信息量的爆炸性增长,如何高效地处理和存储海量数据成为了一个亟待解决的问题。传统的集中式搜索引擎在处理大规模数据时面临诸多挑战,如性能瓶颈...

    Windows平台下Hadoop的Map/Reduce开发

    在Windows平台上进行Hadoop的Map/Reduce开发可能会比在Linux环境下多一些挑战,但通过详细的步骤和理解Map/Reduce的工作机制,开发者可以有效地克服这些困难。以下是对标题和描述中涉及知识点的详细说明: **Hadoop...

    hadoop中map/reduce

    Reduce任务会按照中间键值对的键进行排序,然后把这些键及其对应的值分组,传递给用户定义的Reducer函数。Reducer函数负责对每个键的所有值进行聚合运算,生成最终的结果。 在Hadoop中,MapReduce的工作流程还涉及...

    远程调用执行Hadoop Map/Reduce

    本篇文章将深入探讨“远程调用执行Hadoop Map/Reduce”的概念、原理及其实现过程,同时结合标签“源码”和“工具”,我们将涉及到如何通过编程接口与Hadoop集群进行交互。 Hadoop MapReduce是一种编程模型,用于大...

    对云计算中几种基础设施(Dynamo,Bigtable,Map/Reduce等)的朴素看法

    本文将深入探讨云计算的三大关键技术:Dynamo、Bigtable和Map/Reduce,并对比分析它们的设计理念和应用场景。 首先,Dynamo是亚马逊公司开发的一种分布式键值存储系统,主要用于支持大规模的在线服务,如S3存储服务...

    GEE教程(Map/Reduce 的并行编程概念).ipynb

    该模块介绍了使用 Map/Reduce 的并行编程概念--这是有效使用地球引擎分析大量数据的关键。您将学习如何使用地球引擎 API 计算各种光谱指数,进行云遮蔽,然后使用 Map/reduce 将这些计算应用于图像集合。您还将学习...

    win7_64eclispe插件

    win7_64eclispe插件 解决An internal error occurred during: "Map/Reduce location status updater". org/codehaus/jackson/map/JsonMappingException 重新编译包

    在solr文献检索中用map/reduce

    标题中的“在solr文献检索中用map/reduce”指的是使用Apache Solr,一个流行的开源搜索引擎,结合Hadoop的MapReduce框架来处理大规模的分布式搜索任务。MapReduce是一种编程模型,用于处理和生成大型数据集,它将...

    map/reduce template

    标题中的“map/reduce template”指的是MapReduce编程模型的一个模板或框架,它是Apache Hadoop项目的核心部分,用于处理和生成大数据集。MapReduce的工作原理分为两个主要阶段:Map阶段和Reduce阶段,它允许程序员...

    hadoop教程

    Reducer 是 Map/Reduce 框架中的另一个核心组件,负责将 Mapper 的输出进行排序和聚合,以产生最终的输出结果。 Partitioner 是 Map/Reduce 框架中的一个重要组件,负责将 Mapper 的输出进行分区,以便 Reduce 任务...

    Python核心技术进阶训练篇

    - **Reduce**: reduce() 函数属于 functools 模块,用于累积地对序列中的元素执行某个函数操作。 **3. 实战案例** 例如,我们可以使用lambda表达式配合map()函数来实现列表元素的平方操作: ```python numbers = ...

    基于Map_Reduce的分布式搜索引擎研究

    在对Map/Reduce算法进行分析的基础上,利用开源Hadoop软件设计出高容错高性能的分布式搜索引擎,以面对搜索引擎对海量数据的处理和存储问题

    动态在线Map/Reduce流数据处理模型及作业拓扑管理协议

    针对海量流数据的在线处理需求,提出一种不同于传统Map/Reduce流数据处理的系统模型Flexible workflow.该模型对workflow处理单元进行在线Map/Reduce并行化,实现了SPATE系统;同时为该系统定义一组关于作业的建立、管理...

    Ruby中的类Google Map/Reduce框架Skynet介绍

    不过本文的Skynet没这么恐怖,它是一个ruby版本的Google Map/Reduce框架的名字而已。 Google的Map/Reduce框架实在太有名气了,他可以把一个任务切分为很多份,交给n台计算机并行执行,返回的结果再并行的归并,最后...

    现有student.txt和student-score.txt 将两个文件上传到hdfs上 使用Map/Reduce框架完成下面

    它由两部分组成:`Map`阶段和`Reduce`阶段。MapReduce框架负责调度任务、管理计算节点和处理系统故障等底层细节。 ### MapReduce实现案例分析 根据题目要求,我们需要连接`student.txt`和`student_score.txt`这两...

    基于Java和mapreduce实现的贝叶斯文本分类器设计.zip

    本项目为一个Hadoop课程设计,使用Java语言和map/reduce实现贝叶斯文本分类器。项目的具体内容如下:1:用MapReduce算法实现贝叶斯分类器的训练过程,并输出训练模型; 2:用输出的模型对测试集文档进行分类测试。...

Global site tag (gtag.js) - Google Analytics