离线轻量级大数据平台Spark之JavaRDD关联join操作

对两个RDD进行关联操作，如：

1）文件post_data.txt包含：post_id\title\content

2）文件train.txt包含：dev_id\post_id\praise\time

通过post_id关联，提取post_id\content\praise字段，文件字段用不定长多个空格分割；

参考代码如下：

package scs.contest;import java.util.List;
import java.util.Properties;import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.PairFunction;import scala.Tuple2;public class pageview {public static String path="/home/spark/data";public static void main(String[] args) {//第一步：设置环境字符集，避免中文乱码Properties pps=System.getProperties(); pps.setProperty("file.encoding","UTF-8"); //第二步：txt数据导入并分割成到训练集和预测集SparkConf sparkConf = new SparkConf().setAppName("pageview");JavaSparkContext jsc = new JavaSparkContext(sparkConf);JavaRDD<String> postdataLines =jsc.textFile(path+"/post_data.txt");//id\title\contentJavaRDD<String> trainLines =jsc.textFile(path+"/train.txt");//dev_id\post_id\praise\timeJavaPairRDD<String,String> pdLines=postdataLines.mapToPair(new PairFunction<String,String,String>(){public Tuple2<String, String> call(String t) throws Exception {String[] st=t.split("\\s+");//多个空格情况的分割String content="";for(int i=1;i<st.length;i++){content+=st[i];}return new Tuple2(st[0],content);//title+content都取，title一般为空的情况下}});JavaPairRDD<String,String> ptLines=trainLines.mapToPair(new PairFunction<String,String,String>(){public Tuple2<String, String> call(String t) throws Exception {String[] st=t.split("\\s+");return new Tuple2(st[1], st[2]);}});//合并两个JavaPairRDD，得到训练集和预测集JavaPairRDD<String, Tuple2<String, String>> tLines=pdLines.join(ptLines);//训练街Tuple2<String, Tuple2<String, String>>  tpfirst=tLines.first();System.out.println(tpfirst._1+"|"+tpfirst._2()._1+"|"+tpfirst._2()._2);//预测集，Optional<String>=Optional.empty的记录数JavaPairRDD<String, Tuple2<String, Optional<String>>> pLines=pdLines.leftOuterJoin(ptLines); Tuple2<String, Tuple2<String, String>>  tpfirst1=tLines.first();System.out.println(tpfirst1._1+"|"+tpfirst1._2()._1+"|"+tpfirst1._2()._2);    }
}

执行结果：

507298|无聊的一天又开始了…|1
547732|人生不可能总一帆风顺，总会有一些挫折的。|Optional.empty

参考官网的JavaPariRDD的join操作。

离线轻量级大数据平台Spark之JavaRDD关联join操作相关推荐

离线轻量级大数据平台Spark之MLib机器学习库概念学习
Mlib机器学习库 1.1机器学习概念机器学习有很多定义,倾向于下面这个定义.机器学习是对能通过经验自动改进的计算机算法的研究.机器学习依赖数据经验并评估和优化算法所运行出的模型.机器学习算法尝试根 ...
离线轻量级大数据平台Spark之MLib机器学习协同过滤ALS实例
1.协同过滤协同过滤(Collaborative Filtering,简称CF,WIKI上的定义是:简单来说是利用某个兴趣相投.拥有共同经验之群体的喜好来推荐感兴趣的资讯给使用者,个人透过合作的机制 ...
离线轻量级大数据平台Spark之MLib机器学习库朴素贝叶斯实例
1.朴素贝叶斯介绍表示事件B已经发生的前提下,事件A发生的概率,叫做事件B发生下事件A的条件概率,公式为: 贝叶斯定理: 从已知P(A|B)获得P(B|A)值. 假设A和B代表两类互相影响的事件,如 ...
离线轻量级大数据平台Spark之MLib机器学习库线性回归实例
1.线性回归线性回归是利用称为线性回归方程的函数对一个或多个自变量和因变量之间关系进行建模的一种回归分析方法,只有一个自变量的情况称为简单回归,大于一个自变量情况的叫做多元回归,在实际情况中大多数都 ...
离线轻量级大数据平台Spark之MLib机器学习库聚类算法KMeans实例
1.KMeans算法所谓聚类问题,就是给定一个元素集合D,其中每个元素具有n个可观察属性,使用某种算法将D划分成k个子集,要求每个子集内部的元素之间相异度尽可能低,而不同子集的元素相异度尽可能高.其 ...
离线轻量级大数据平台Spark之单机部署及Java开发
1.Spark平台基本介绍 Spark 由加州大学伯克利分校 AMP 实验室 (Algorithms, Machines, and PeopleLab) 开发,可用来构建大型的.低延迟的数据分析应用程 ...
离线轻量级大数据平台Spark之中文字符显示问题的解决
问题:spark读取文本文件转化成JavaRDD后发现中文字符显示乱码. 在spark-shell环境里执行:System.getProperty("file.encoding") ...
离线轻量级大数据平台Spark之读取CSV文件实例
Spark的RDD数据集很适合处理轻量文件,一般场景下是excel文件,可以将excel文件另存为CSV(逗号分隔),Spark读取CSV文件形成RDD. 1.序列化类Record,用于保存字段 pa ...
离线轻量级大数据平台Spark之MLib机器学习库Word2Vec实例
Word2Vecword2vec能将文本中出现的词向量化,可以在捕捉语境信息的同时压缩数据规模.Word2Vec实际上是两种不同的方法:Continuous Bag of Words (CBOW) 和 ...

离线轻量级大数据平台Spark之JavaRDD关联join操作

离线轻量级大数据平台Spark之JavaRDD关联join操作相关推荐

最新文章

热门文章