對兩個RDD進行關(guān)聯(lián)操作，如：

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

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

通過post_id關(guān)聯(lián)，提取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) {//第一步：設(shè)置環(huán)境字符集，避免中文亂碼Properties pps=System.getProperties(); pps.setProperty("file.encoding","UTF-8"); //第二步：txt數(shù)據(jù)導(dǎo)入并分割成到訓(xùn)練集和預(yù)測集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，得到訓(xùn)練集和預(yù)測集JavaPairRDD<String, Tuple2<String, String>> tLines=pdLines.join(ptLines);//訓(xùn)練街Tuple2<String, Tuple2<String, String>> tpfirst=tLines.first();System.out.println(tpfirst._1+"|"+tpfirst._2()._1+"|"+tpfirst._2()._2);//預(yù)測集，Optional<String>=Optional.empty的記錄數(shù)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); } }
執(zhí)行結(jié)果：

507298|無聊的一天又開始了…|1 547732|人生不可能總一帆風(fēng)順，總會有一些挫折的。|Optional.empty

參考官網(wǎng)的JavaPariRDD的join操作。

總結(jié)

以上是生活随笔為你收集整理的离线轻量级大数据平台Spark之JavaRDD关联join操作的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。