spark join算子

java

/** 
 *join算子是根据两个rdd的key进行关联操作,类似scala中的拉链操作,返回的新元素为<key,value>,一对一
 *@author Tele
 *
 */
public class JoinDemo {
    private static SparkConf conf = new SparkConf().setMaster("local").setAppName("joindemo");
    private static JavaSparkContext jsc = new JavaSparkContext(conf);
    public static void main(String[] args) {
        
        //假设每个学生只有一门成绩
        List<Tuple2<Integer,String>> studentList = Arrays.asList(
                                                    new Tuple2<Integer,String>(1,"tele"),
                                                    new Tuple2<Integer,String>(2,"yeye"), 
                                                    new Tuple2<Integer,String>(3,"wyc")
                                                    );
        
        List<Tuple2<Integer,Integer>> scoreList = Arrays.asList(
                                                  new Tuple2<Integer,Integer>(1,100),
                                                  new Tuple2<Integer,Integer>(1,1100),
                                                  new Tuple2<Integer,Integer>(2,90),
                                                  new Tuple2<Integer,Integer>(3,70)
                                                  );
                
        
        JavaPairRDD<Integer, String> studentRDD = jsc.parallelizePairs(studentList);
        JavaPairRDD<Integer, Integer> scoreRDD = jsc.parallelizePairs(scoreList);
        
        //注意此处生成的新rdd对的参数类型,第一个泛型参数为key的类型,Tuple2的String与Integer分别对应原rdd的value类型
        JavaPairRDD<Integer, Tuple2<String, Integer>> result = studentRDD.join(scoreRDD);
        
        result.foreach(new VoidFunction<Tuple2<Integer,Tuple2<String,Integer>>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public void call(Tuple2<Integer, Tuple2<String, Integer>> t) throws Exception {
                System.out.println("学号:" + t._1);
                System.out.println("姓名:" + t._2._1);
                System.out.println("成绩:" + t._2._2);
                System.out.println("=================");
            }
        });
        
        jsc.close();
        
    }
}

scala

object JoinDemo {
    def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setMaster("local").setAppName("joindemo");     
      val sc = new SparkContext(conf);
      
      val studentArr = Array((1,"tele"),(2,"yeye"),(3,"wyc"));
      val scoreArr = Array((1,100),(2,80),(3,100));
      
      val studentRDD = sc.parallelize(studentArr,1);
      val scoreRDD = sc.parallelize(scoreArr,1);
      
      val result = studentRDD.join(scoreRDD);
      
      result.foreach(t=>{
        println("学号:" + t._1);
        println("姓名:" + t._2._1);
        println("成绩:" + t._2._2);
        println("============")
      })
      
    }
}