快捷導(dǎo)航

利用hadoop查詢兩兩之間有共同好友及他倆的共同好友都是誰(shuí)

更新時(shí)間：2024年10月10日 11:26:58 作者：云游遍天下

一想到要實(shí)現(xiàn)求共同好友的功能，很多人都會(huì)想到redis來(lái)實(shí)現(xiàn)。但是redis存儲(chǔ)和數(shù)據(jù)和計(jì)算時(shí)需要耗費(fèi)較多的內(nèi)存資源。所以文本將介紹另一種方法，即利用Hadoop中的MapReduce來(lái)實(shí)現(xiàn)，感興趣的可以了解一下

編碼思路：

第一步是可以把好友當(dāng)作key，value是擁有key好友的用戶，例如:擁有好友B的是：A,F,J,E用戶

第二步在第一步結(jié)果后，雙重for循環(huán)進(jìn)行兩兩之間進(jìn)行拼接，這樣就可以得出正確結(jié)果

具體代碼實(shí)現(xiàn)：

第一步：

package com.zsy.mr.commonfriend;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.StringUtils;
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.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;

public class commonFriendStepOne {
	static class commonFriendStepOneMapper extends Mapper<LongWritable, Text, Text, Text>{
		Text k = new Text();
		Text v = new Text();
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			//通過(guò)過(guò)冒號(hào)分割
			String[] splits = value.toString().split(":");
			//獲取擁有好友的用戶名
			String name = splits[0];
			//獲取該用戶下的好友列表
			String[] friends = StringUtils.isNotBlank(splits[1])?  splits[1].split(","):null;
			if(friends != null) {
				//循環(huán)好友，好友當(dāng)作key，擁有好友的用戶名當(dāng)作value
				for (String friend : friends) {
					k.set(friend);
					v.set(name);
					context.write(k, v);
				}
			}
		}
	}
	
	static class commonFriendStepOneReducer extends Reducer<Text, Text, Text, Text>{
		Text v = new Text();
		@Override
		protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			List<String> resultList = new ArrayList<String>();//實(shí)際生產(chǎn)代碼不建議用list接收，應(yīng)該是直接處理掉
			//處理數(shù)據(jù)，該數(shù)據(jù)是擁有key好友的所有用戶
			for (Text value : values) {
				resultList.add(value.toString());
			}
			v.set(StringUtils.join(resultList, ","));
			context.write(key, v);
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		/*conf.set("mapreduce.framework.name", "yarn");
        conf.set("yarn.resoucemanger.hostname", "hadoop01");*/
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(commonFriendStepOne.class);
		
		//指定本業(yè)務(wù)job要使用的業(yè)務(wù)類
		job.setMapperClass(commonFriendStepOneMapper.class);
		job.setReducerClass(commonFriendStepOneReducer.class);
		
		//指定mapper輸出的k v類型  如果map的輸出和reduce的輸出一樣，只需要設(shè)置輸出即可
		//job.setMapOutputKeyClass(Text.class);
		//job.setMapOutputValueClass(IntWritable.class);
		
		//指定最終輸出kv類型（reduce輸出類型）
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		//指定job的輸入文件所在目錄
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		//指定job的輸出結(jié)果目錄
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		//將job中配置的相關(guān)參數(shù)，以及job所有的java類所在 的jar包，提交給yarn去運(yùn)行
		//job.submit();無(wú)結(jié)果返回，建議不使用它
		boolean res = job.waitForCompletion(true);
		
		System.exit(res?0:1);
	}
}

結(jié)果：

第二步：

代碼實(shí)現(xiàn)

package com.zsy.mr.commonfriend;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
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;

public class commonFriendStepTwo {

	static class commonFriendStepTwoMapper extends Mapper<LongWritable, Text, Text, Text>{
		Text k = new Text();
		Text v = new Text();
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			String[] splits = value.toString().split("\t");
			//獲取好友
			String friend = splits[0];
			//獲取擁有該好友所有的用戶信息
			String[] names = splits[1].split(",");
			//進(jìn)行排序，防止計(jì)算數(shù)據(jù)重復(fù)，例如：A-B和B-A其實(shí)一個(gè)對(duì)
			Arrays.sort(names);
			//進(jìn)行雙重for循環(huán)
			for (int i = 0; i < names.length-1; i++) {
				String string = names[i];
				for (int j = i+1; j < names.length; j++) {
					String string2 = names[j];
					k.set(string+"-"+string2);
					v.set(friend);
					context.write(k, v);
				}
			}
		}
	}
	
	static class commonFriendStepTwoReducer extends Reducer<Text, Text, Text, NullWritable>{
		Text k = new Text();
		@Override
		protected void reduce(Text key, Iterable<Text> value, Reducer<Text, Text, Text, NullWritable>.Context context)
				throws IOException, InterruptedException {
			List<String> resultList = new ArrayList<String>();//實(shí)際生產(chǎn)代碼不建議用list接收，應(yīng)該是直接處理掉
			for (Text text : value) {
				resultList.add(text.toString());
			}
			k.set(key.toString()+":"+ StringUtils.join(resultList,","));
			context.write(k, NullWritable.get());
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		/*conf.set("mapreduce.framework.name", "yarn");
        conf.set("yarn.resoucemanger.hostname", "hadoop01");*/
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(commonFriendStepTwo.class);
		
		//指定本業(yè)務(wù)job要使用的業(yè)務(wù)類
		job.setMapperClass(commonFriendStepTwoMapper.class);
		job.setReducerClass(commonFriendStepTwoReducer.class);
		
		//指定mapper輸出的k v類型  如果map的輸出和reduce的輸出一樣，只需要設(shè)置輸出即可
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		//指定最終輸出kv類型（reduce輸出類型）
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		
		//指定job的輸入文件所在目錄
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		//指定job的輸出結(jié)果目錄
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		//將job中配置的相關(guān)參數(shù)，以及job所有的java類所在 的jar包，提交給yarn去運(yùn)行
		//job.submit();無(wú)結(jié)果返回，建議不使用它
		boolean res = job.waitForCompletion(true);
		
		System.exit(res?0:1);
	}
}

結(jié)果：