文档章节

Hadoop中的分组(15)

肖鋭
 肖鋭
发布于 2014/03/04 21:36
字数 540
阅读 38
收藏 0
package grounp;

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	2
   3	3
 * @author Xr
 *
 */
public class groupApp {
	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, groupApp.class.getName());
		
		FileInputFormat.setInputPaths(job, INPUT_PATH);
		job.setMapperClass(MyMapper.class);
		//自定义键
		job.setMapOutputKeyClass(NewKey.class);
		job.setMapOutputValueClass(LongWritable.class);
		//自定义分组
		job.setGroupingComparatorClass(NewGroupCompator.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;
			}
		}
	}
}
class NewGroupCompator implements RawComparator<NewKey>{

	@Override
	public int compare(NewKey o1, NewKey o2) {
		return 0;
	}
	
	/**
	 * 比较字节数组中指定的字节序列的大小
	 * @param b1	第一个参与比较的字节数组
	 * @param s1	第一个参与比较的字节数组的开始位置
	 * @param l1	第一个参与比较的字节数组的字节长度
	 * @param b2	第二个参与比较的字节数组	
	 * @param s2	第二个参与比较的字节数组的开始位置
	 * @param l2	第二个参与比较的字节数组的字节长度
	 * @return
	 */
	@Override
	public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
		return WritableComparator.compareBytes(b1, s1, 8, b2, s2, 8);
	}
}

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


© 著作权归作者所有

肖鋭
粉丝 10
博文 62
码字总数 29531
作品 0
朝阳
程序员
私信 提问
Apache Sentry 第一弹:Server启动、连接Hue、分组详解

前言: Sentry是Hadoop安全方面的一个开源组件,目前还在孵化中,地址:https://sentry.incubator.apache.org。 原属于Cloudera开发,后来贡献给了Apache。关于它的强大直接摘录一段,全文请...

王二铁
2015/08/21
3.9K
0
Hadoop Mapreduce分区、分组、二次排序过程详解

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

张欢19933
2016/04/07
525
0
【hadoop】10.HDFS-机架感知

简介 通过本教程您可以学习到: 网络拓扑的概念 节点距离的计算 机架感知 如何修改默认的机架设置策略 验证机架感知 1、网络相关知识 1.1、网络拓扑 为了接下来对一些内部机制有所了解,我们先...

Areya
2019/01/12
58
0
Twitter Storm Stream Grouping编写自定义分组实现

自定义Grouping测试 Storm是支持自定义分组的,本篇文章就是探究Storm如何编写一个自定义分组器,以及对Storm分组器如何分组数据的理解。 这是我写的一个自定义分组,总是把数据分到第一个T...

震秦
2014/04/17
798
0
MapReduce的自制Writable分组输出及组内排序

问题描述: 输入文件格式如下: name1 2 name3 4 name1 6 name1 1 name3 3 name1 0 要求输出的文件格式如下: name1 0,1,2,6 name3 3,4 要求是按照第一列分组,name1与name3也是按照顺序...

zhao_xiao_long
2013/09/03
0
0

没有更多内容

加载失败,请刷新页面

加载更多

如何优雅地检测JavaScript中的空闲时间?

是否可以在JavaScript中检测“ 空闲 ”时间? 我的主要用例可能是预取或预加载内容。 空闲时间: 用户不活动或没有使用CPU的时间 #1楼 如果不使用jQuery,则仅使用普通JavaScript: var inac...

技术盛宴
40分钟前
31
0
获取枚举值的属性

我想知道是否可以获取枚举值而不是枚举本身的属性? 例如,假设我有以下枚举: using System.ComponentModel; // for DescriptionAttributeenum FunkyAttributesEnum{ [Description(...

javail
今天
78
0
concurrently 启动多个serve时命令行输出混乱

在script中配置中加入参数 -r "start": "npx concurrently -r \"npm:serve\" \"npm:web\"" 输出比较整洁 输出混乱, 会将有用的信息冲掉...

阿豪boy
今天
48
0
每天AC系列(三):电话号码的字母组合

1 题目 Leetcode第17题。 数字2-9映射字母,给出一个包含字符串的数字,列出字母的所有组合。 2 递归 拿到这样的题目想到了递归实现,创建一个Map映射对应的字母,然后把它传给递归函数,同时...

Blueeeeeee
今天
44
0
Kettle自定义jar包供javascript使用

我们都知道 Kettle 是用 Java 语言开发,并且可以在 JavaScript 里面直接调用 java 类方法。所以有些时候,我们可以自定义一些方法,来供 JavaScript 使用。 本篇文章有参考自:https://www...

CREATE_17
昨天
114
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部