写在前面

你们好我是啊晨，一个大数据分享者兼一个努力成为大垃圾的小垃圾
本章介绍，使用spark计算共同好友，相信看这篇文章之前都有了解做过MapReduce的共同好友，文章后会有MapReduce的方法，大家自行比较一下哈。
如有其它需要请阅读我的其它大数据文章，谢谢
中间有什么问题请留言，请珍惜现在的时间：

描述

如网站有如下关系数据：
friends.txt

A:B,C,D,F,E,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:A,B,C,D,E,O,M
G:A,C,D,E,F
H:A,C,D,E,O
I:A,O
J:B,O
K:A,C,D
L:D,E,F
M:E,F,G
O:A,H,I,J

数据说明：
A:B,C,D,F,E,O
每行数据以冒号为分隔符：
1.冒号左边是网站的一个用户A；
2.冒号右边是网站A的好友列表（各好友间以逗号为分隔符）；

现在需要对网站的用户进行分析，找出那些用户两两之间有共同好友，以及他俩的共同好友都有那些人。
如：A、B两个用户拥有共同好友C和E；

(F-H,D,O,A,C,E)
(A-F,B,D,O,C,E)
(B-H,A,C,E)
(F-I,O,A)
(G-O,A)
(B-C,A)
(D-F,A,E)
(B-M,E)
(C-G,F,D,A)
....

计算

完整代码如下：

package sparkimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}//共同好友聚合
object CommonFriend {def main(args: Array[String]): Unit = {//confval conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getName)//scval sc = new SparkContext(conf)//读取数据val source: RDD[String] = sc.textFile("file:///E:\\work\\friends.txt")//处理数据val friendAndUser: RDD[(String, List[String])] = source.flatMap(line => {//按照：进行切分val infos: Array[String] = line.split(":", -1)//获取用户val user = infos(0)//获取好友列表val friends = infos(1).split(",", -1)//好友为k，用户作为vfriends.map(friend => (friend, List(user)))})// 好友，用户,聚合，如第一次mapreduce完成val friendAndUsers = friendAndUser.reduceByKey((list1, list2) => list1 ++ list2)val userAndUserAndFriend: RDD[(String, String)] = friendAndUsers.flatMap(tp => {//为用户正序排序,避免重复获取val list: List[String] = tp._2.sortBy(x => x)//创建容器存放用户--用户，好友var listFriend = List[(String, String)]()//遍历用户for (i <- 0 until list.size - 1) {for (j <- i + 1 until list.size) {listFriend :+= (list(i) + "-" + list(j), tp._1)}}listFriend})userAndUserAndFriend.reduceByKey((x, y) => x.concat(",").concat(y)).foreach(println)//关闭资源sc.stop()}
}

MapReduce计算共同好友

参考文章https://zhuanlan.zhihu.com/p/50236955

job1的mapper类

map函数中的逻辑相对简单，只需要对原始数据按分隔符切分，然后将“好友”作为key，用户作为value输出即可。

public static class CommonFriendStep1Mapper extends Mapper<LongWritable, Text, Text, Text>{// 输入数据形式如：      A:B,C,D,F,E,O@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {// 将maptask所传入的一行数据按照冒号切分String[] split = value.toString().split(":");// 得到数据中的用户String user = split[0];// 得到数据中的"好友们"String[] friends = split[1].split(",");// 将每一个"好友"作为key，用户作为value，返回给maptaskfor (String f : friends) {context.write(new Text(f), new Text(user));}}
}

job1的Reducer类

reduce函数中的核心处理逻辑是对拥有某个共同好友的所有用户两两拼对，然后输出各种两两对组合及他们所拥有的共同好友。

public static class CommonFriendStep1Reducer extends Reducer<Text, Text, Text, Text>{// f:某个"好友"// users:拥有这个f好友的一堆用户@Overrideprotected void reduce(Text f, Iterable<Text> users, Context context) throws IOException, InterruptedException {ArrayList<Text> userList = new ArrayList<>();// 将这一组拥有共同好友f的user们从迭代器中取出，放入一个arraylist暂存for (Text u : users) {userList.add(new Text(u));}// 对users排个序，以免拼俩俩对时出现A-F 又有F-A的现象Collections.sort(userList);// 把这一对user进行两两组合，并将://1.组合作为key//2.共同的好友f作为value//返回给reduce task作为本job的最终结果for(int i=0;i<userList.size()-1;i++) {for(int j=i+1;j<userList.size();j++) {// 输出 "用户-用户" 两两对，及他俩的共同好友context.write(new Text(userList.get(i)+"-"+userList.get(j)), f);}}}
}

job1的客户端

public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(CommonFriendStep1.class);// 设置job的mapper类job.setMapperClass(CommonFriendStep1Mapper.class);// 设置job的reducer类job.setReducerClass(CommonFriendStep1Reducer.class);// 设置map阶段输出的key：value数据类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);// 设置reduce阶段输出的key：value数据类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);// 判断结果输出路径是否已存在，如果已经存在，则删除。以免在测试阶段需要反复手动删除输出目录FileSystem fs = FileSystem.get(conf);Path out = new Path(args[0]);if(fs.exists(out)) {fs.delete(out, true);}// 设置数据输入输出路径FileInputFormat.setInputPaths(job, new Path(args[1]));FileOutputFormat.setOutputPath(job,out);// 提交job给yarn或者local runner来运行job.waitForCompletion(true);}

job2的Mapper类

map()方法逻辑相对简单，只需要对上一个步骤所产生的数据切分，然后以“两两对”作为key，他们共同的好友作为value输出即可。

public static class CommonFriendStep2Mapper extends Mapper<LongWritable, Text, Text, Text>{// 上一个job所产生的数据是本次job读取的数据： B-C  A@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 将数据按制表符切分String[] split = value.toString().split("\t");// 将切出来的B-C用户对作为key，共同好友A作为value// 返回给map taskcontext.write(new Text(split[0]), new Text(split[1]));}
}

job2的Reducer类

reduce()方法会获得“两两对”用户组合所拥有的所有好友value，从而只需要迭代每一组value进行字符串拼接，即可得到最终结果

public static class CommonFriendStep2Reducer extends Reducer<Text, Text, Text, Text>{@Overrideprotected void reduce(Text pair, Iterable<Text> friends, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {// 构造一个StringBuilder用于拼接字符串StringBuilder sb = new StringBuilder();// 将这个用户对的所有共同好友拼接在一起for (Text f : friends) {sb.append(f).append(",");}// 将用户对作为key，拼接好的所有共同好友作为value，返回给reduce taskcontext.write(pair, new Text(sb.substring(0, sb.length()-1)));}
}

job2的客户端

public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(CommonFriendStep2.class);// 设置job的mapper类和reducer类job.setMapperClass(CommonFriendStep2Mapper.class);job.setReducerClass(CommonFriendStep2Reducer.class);// 设置map阶段输出key:value数据的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);// 设置reudce阶段输出key:value数据的类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);// 检测输出目录是否已存在，如果已存在则删除，以免在测试阶段需要反复手动删除输出目录FileSystem fs = FileSystem.get(conf);Path out = new Path("F:\\mrdata\\friends\\out-2");if(fs.exists(out)) {fs.delete(out, true);}// 设置数据输入输出目录FileInputFormat.setInputPaths(job, new Path("F:\\mrdata\\friends\\out-1"));FileOutputFormat.setOutputPath(job,out);// 提交job到yarn或者local runner执行job.waitForCompletion(true);}

本篇完结，Spark比MR少了超级多代码，很是舒服，一定敲敲敲着去理解，下篇继续更新大数据其他内容，记得关注点赞支持哈，谢谢观看

单是说不行，要紧的是做。——鲁迅

如何使用Spark计算共同好友？相关推荐

spark 算子例子_10年大数据架构师，用一文带你玩转Spark计算框架，你能读懂吗？...
首先明确一点:学计算框架主要就是学2部分: 1.资源调度 2.任务调度写一个spark程序包含加载配置文件,创建上下文,创建RDD , 调用RDD的算子,用户在算子中自定义的函数 map端:狭窄的理 ...
BigData之Spark：Spark计算引擎的简介、下载、经典案例之详细攻略
BigData之Spark:Spark计算引擎的简介.下载.经典案例之详细攻略目录 Spark的简介 1.Spark三大特点 Spark的下载 Spark的经典案例 1.Word Count 2.P ...
Java进行spark计算
首先在Linux环境安装spark: 可以从如下地址下载最新版本的spark: https://spark.apache.org/downloads.html 这个下载下来后是个tgz的压缩包,解压后 ...
spark求共同好友
使用spark求共同好友: 在做一些项目时,又时可能遇到一些类似与求共同好友的要求,可以根据共同好友进行推荐添加好友,就比如说A和B的好友有M,K,O,L,那么有可能A和B就是也认识,所以可以把B推荐 ...
Spark计算工具类
Vector vectors.txt 1 2.3 4.5 3 3.1 5.6 4 3.2 7.8 处理vectors.txt文件RDD[String]->RDD[Vector] package ...
使用spark计算文档相似度
2019独角兽企业重金招聘Python工程师标准>>> 1.TF-IDF文档转换为向量以下边三个句子为例罗湖发布大梧桐新兴产业带整体规划深化伙伴关系,增强发展动力为世界经济发 ...
spark计算操作整理
spark 的计算流程大概如图: 其中, 通过多次处理, 生成多个中间数据, 最后对结果进行操作获得数据. 本文不涉及任何原理, 仅总结spark在处理的时候支持的所有操作, 方便后面使用的时候, 可 ...
降低 Spark 计算成本 50.18 %，使用 Kyligence 湖仓引擎构建云原生大数据底座，为计算提速 2x
2023 中国开源未来发展峰会于 5 月 13 日成功举办.在大会开源原生商业分论坛,Kyligence 解决方案架构高级总监张小龙发表<云原生大数据底座演进 >主题演讲,向与会嘉宾介绍了 ...
spark常用功能：使用Spark计算数列统计值
参考 : -- https://cloud.tencent.com/developer/article/1475487 先来回顾一下数据和对应的统计结果: 本文使用的是iris分类数据集,数据下载地址 ...

如何使用Spark计算共同好友？

文章目录