建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
Hadoop中如何實現(xiàn)分組
這篇文章主要為大家展示了“Hadoop中如何實現(xiàn)分組”，內(nèi)容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領(lǐng)大家一起研究并學(xué)習(xí)一下“Hadoop中如何實現(xiàn)分組”這篇文章吧。
為徽縣等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計制作服務(wù)，及徽縣網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為成都網(wǎng)站設(shè)計、網(wǎng)站建設(shè)、徽縣網(wǎng)站設(shè)計，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會得到認(rèn)可，從而選擇與我們長期合作。這樣，我們也可以走得更遠(yuǎn)！
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{
@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{
@Override
protected void reduce(NewKey key2, Iterable 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{
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{
@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)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！
網(wǎng)頁標(biāo)題：Hadoop中如何實現(xiàn)分組
路徑分享：http://fisionsoft.com.cn/article/psocpg.html
