文档章节

Hadoop中的自定义Map端输出Key(14)

肖鋭
 肖鋭
发布于 2014/03/04 21:32
字数 386
阅读 87
收藏 0
package sort;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/** 
 * 自定义排序
 * 初始结果:
 * 3	3
 * 3	2
 * 3	1
 * 2	2
 * 2	1
 * 1	1
 * 输出结果:
   1	1
   2	1
   2	2
   3	1
   3	2
   3	3
 * @author Xr
 *
 */
public class SortApp {
	public static final String INPUT_PATH = "hdfs://hadoop:9000/data";
	public static final String OUTPUT_PATH = "hdfs://hadoop:9000/datas";
	public static void main(String[] args)throws Exception{
		Configuration conf = new Configuration();
		existsFile(conf);
		Job job = new Job(conf, SortApp.class.getName());
		
		FileInputFormat.setInputPaths(job, INPUT_PATH);
		job.setMapperClass(MyMapper.class);
		//自定义输出Key
		job.setMapOutputKeyClass(NewKey.class);
		job.setMapOutputValueClass(LongWritable.class);
		job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongWritable.class);
		FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
		job.waitForCompletion(true);
	}
	private static void existsFile(Configuration conf) throws IOException,
			URISyntaxException {
		FileSystem fs = FileSystem.get(new URI(OUTPUT_PATH),conf);
		if(fs.exists(new Path(OUTPUT_PATH))){
			fs.delete(new Path(OUTPUT_PATH),true);
		}
	}
}
class MyMapper extends Mapper<LongWritable, Text, NewKey, LongWritable>{

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String string = value.toString();
		String[] split = string.split("\t");
		NewKey k2 = new NewKey();
		k2.set(Long.parseLong(split[0]),Long.parseLong(split[1]));
		context.write(k2, new LongWritable(Long.parseLong(split[1])));
	}
}
class MyReducer extends Reducer<NewKey, LongWritable, LongWritable, LongWritable>{

	@Override
	protected void reduce(NewKey key2, Iterable<LongWritable> values,Context context)
			throws IOException, InterruptedException {
		long max = Long.MIN_VALUE;
		for(LongWritable v2 : values){
			long l = v2.get();
			if(l>max){
				max = l;
			}
		}
		context.write(new LongWritable(key2.first),new LongWritable(max));
	}
} 
class NewKey implements WritableComparable<NewKey>{
	long first;
	long second;
	
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeLong(this.first);
		out.writeLong(this.second);
	}

	public void set(long parseLong, long parseLong2) {
		this.first = parseLong;
		this.second = parseLong2;
	}

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

	@Override
	public int compareTo(NewKey o) {
		if(this.first==o.first){
			if(this.second < o.second){
				return -1;
			}else if(this.second == o.second){
				return 0;
			}else{
				return 1;
			}
		}else{
			if(this.first < o.first){
				return -1;
			}else{
				return 1;
			}
		}
	}
}


                                                                  Name:Xr
                                                                  Date:2014-03-04 21:31





© 著作权归作者所有

肖鋭
粉丝 10
博文 62
码字总数 29531
作品 0
朝阳
程序员
私信 提问
hadoop 平台运行mapreduce代码

WordCount是写hadoop mapreduce入门级程序,会写wordcount的话,基本上80%的mapreduce就懂了。 mapreduce分为map过程和reduce过程,用户可以根据自己的业务自定义map过程和reduce过程。 以w...

leiline
2018/01/14
29
0
通过java api提交自定义hadoop 作业

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/qq1010885678/article/details/43734989 通过API操作之前要先了解几个基本知识 一、hadoop的基本数据类型和j...

jchubby
2015/02/11
0
0
Hadoop Mapreduce分区、分组、二次排序过程详解

1、MapReduce中数据流动 (1)最简单的过程: map - reduce (2)定制了partitioner以将map的结果送往指定reducer的过程: map - partition - reduce (3)增加了在本地先进性一次reduce(优...

张欢19933
2016/04/07
525
0
Hadoop之MapReduce自定义二次排序流程实例详解

1.如何解决MapReduce二次排序? 2.Map端如何处理? 3.Reduce端如何处理? 4.MapReduce二次排序是如何具体实现的呢? 一、概述 MapReduce框架对处理结果的输出会根据key值进行默认的排序,这个...

张欢19933
2016/04/07
102
0
Apache Spark探秘:实现Map-side Join和Reduce-side Join

Apache Spark探秘:实现Map-side Join和Reduce-side JoinCategory: Frameworks (Spark) On YARN View: 13,756 Author: Dong 作者:Dong | 新浪微博:西成懂 | 可以转载, 但必须以超链接形式标...

片刻
2015/12/23
218
0

没有更多内容

加载失败,请刷新页面

加载更多

CSS字体边框?

随着所有新的CSS3边框填充( -webkit ,...),现在是否可以为字体添加边框? (就像蓝色的Twitter徽标周围的纯白色边框一样)。 如果不是,是否有任何不太丑陋的骇客会在CSS / XHTML中完成此...

javail
30分钟前
48
0
是DateTime.Now是测量函数性能的最佳方法吗?

我需要找到一个瓶颈,并且需要准确地测量时间。 以下代码段是衡量性能的最佳方法吗? DateTime startTime = DateTime.Now;// Some execution processDateTime endTime = DateTime.Now;...

技术盛宴
45分钟前
31
0
Hive安装

下载上传apache-hive-2.1.1-bin.tar.gz文件并解压 tar -zxvf apache-hive-2.1.1-bin.tar.gz -C /export/servers 配置环境变量 vi /etc/profileexport HIVE_HOME=/export/servers/apache......

长臂猿猴
今天
53
0
Fedora 31 - 荣耀魔法本问题记录

最近荣耀魔法本装了Fedora31系统,没有像Deepin 15.11那样出现亮度不能调节和网卡不能用的问题。但是也有不少小问题: 触摸板有时启动后失效,需要重启。 触摸板的右键失效。 待机后进入桌面...

wffger
今天
46
0
ElasticSearch深入:内部机制浅析(三)@

前言 上篇从分布式的角度阐述了 ES 的分布式设计和思想,这一篇打算与 Lucene 结合起来,摸透一些 ES 的常遇到的概念,我们可以将了解到的这些东西应用到优化实践中去。 一、Shard Shard 实际...

HLee
今天
42
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部