這篇文章主要為大家展示了“Hadoop中如何實現(xiàn)分組”，內(nèi)容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領(lǐng)大家一起研究并學習一下“Hadoop中如何實現(xiàn)分組”這篇文章吧。

為徽縣等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計制作服務，及徽縣網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務為成都網(wǎng)站設(shè)計、網(wǎng)站建設(shè)、徽縣網(wǎng)站設(shè)計，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務。我們深信只要達到每一位用戶的要求，就會得到認可，從而選擇與我們長期合作。這樣，我們也可以走得更遠！

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;
/** 
 * 自定義分組
 * 初始結(jié)果:
 * 3	3
 * 3	2
 * 3	1
 * 2	2
 * 2	1
 * 1	1
 * 輸出結(jié)果：
   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;
	}
	
	/**
	 * 比較字節(jié)數(shù)組中指定的字節(jié)序列的大小
	 * @param b1	第一個參與比較的字節(jié)數(shù)組
	 * @param s1	第一個參與比較的字節(jié)數(shù)組的開始位置
	 * @param l1	第一個參與比較的字節(jié)數(shù)組的字節(jié)長度
	 * @param b2	第二個參與比較的字節(jié)數(shù)組	
	 * @param s2	第二個參與比較的字節(jié)數(shù)組的開始位置
	 * @param l2	第二個參與比較的字節(jié)數(shù)組的字節(jié)長度
	 * @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);
	}
}

以上是“Hadoop中如何實現(xiàn)分組”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學習更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！

本文題目：Hadoop中如何實現(xiàn)分組
當前地址：http://aaarwkj.com/article36/psocpg.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站設(shè)計、微信公眾號、App設(shè)計、搜索引擎優(yōu)化、建站公司、品牌網(wǎng)站制作

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)