如何使用spark Context轉(zhuǎn)成RDD

這篇文章主要介紹“如何使用spark Context轉(zhuǎn)成RDD”，在日常操作中，相信很多人在如何使用spark Context轉(zhuǎn)成RDD問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”如何使用spark Context轉(zhuǎn)成RDD”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

一. 背景

在spark rdd轉(zhuǎn)換算子中join和cogroup是有些需要區(qū)分的算子轉(zhuǎn)換，這里使用示例來(lái)說(shuō)明一下。

二. 示例

1.構(gòu)建List示例數(shù)據(jù)

List<Tuple2<Integer, String>> studentsList = Arrays.asList(
      new Tuple2<Integer,String>(1,"xufengnian"),
      new Tuple2<Integer,String>(2,"xuyao"),
      new Tuple2<Integer,String>(2,"wangchudong"),
      new Tuple2<Integer,String>(3,"laohuang")
      );

List<Tuple2<Integer, Integer>> scoresList = Arrays.asList(
      new Tuple2<Integer,Integer>(1,100),
      new Tuple2<Integer,Integer>(2,90),
      new Tuple2<Integer,Integer>(3,80),
      new Tuple2<Integer,Integer>(1,101),
      new Tuple2<Integer,Integer>(2,91),
      new Tuple2<Integer,Integer>(3,81),
      new Tuple2<Integer,Integer>(3,71)
      );

2.使用sparkContext轉(zhuǎn)成RDD

JavaPairRDD<Integer,String> studentsRDD = sc.parallelizePairs(studentsList);
JavaPairRDD<Integer,Integer> scoresRDD = sc.parallelizePairs(scoresList);

//studentsRDD 為：List<Tuple2<Integer, String>>
//(1,xufengnian)(2,xuyao)(2,wangchudong)(3,laohuang),下面進(jìn)行打印查看

studentsRDD.foreach(new VoidFunction<Tuple2<Integer,String>>(){
   public void call(Tuple2<Integer,String> tuple){
      System.out.println(tuple._1);//1 2 3
      System.out.println(tuple._2);// xufengnian xuyao laohuang
   }
});

3.進(jìn)行join

/*
前面數(shù)據(jù)
(1,xufengnian)(2,xuyao)(2,"wangchudong")(3,laohuang)
(1,100)(2,90)(3,80)(1,101)(2,91)(3,81)(3,71)
join之后：
(1,(xufengnian,100))(1,(xufengnian,101))(3,(laohuang,80))(3,(laohuang,81))(3,(laohuang,71))
(2,(xuyao,90))(2,(xuyao,91))(2,(wangchudong,90))(2,(wangchudong,91))
*/
JavaPairRDD<Integer, Tuple2<String, Integer>> studentScores = studentsRDD.join(scoresRDD);
//join為key相同的join,key不變，value變成(string,integer)
studentScores.foreach(new VoidFunction<Tuple2<Integer,Tuple2<String,Integer>>>() {
   
   private static final long serialVersionUID = 1L;

   @Override
   public void call(Tuple2<Integer, Tuple2<String, Integer>> student)
         throws Exception {
      System.out.println("student id: " + student._1);//1 1 3
      System.out.println("student name: " + student._2._1);//xufengnian xufengnian laohuang
      System.out.println("student score: " + student._2._2);//100 101 80
      System.out.println("===================================");
   }
});

4.進(jìn)行cogroup

/*
前面的數(shù)據(jù)
(1,xufengnian)(2,xuyao)(2,"wangchudong")(3,laohuang)
(1,100)(2,90)(3,80)(1,101)(2,91)(3,81)(3,71)
cogroup之后：
(1,([xufengnian],[100,101]))  (3,([laohuang],[80,81,71]))  (2,([xuyao,wangchudong],[90,91]))
*/
JavaPairRDD<Integer,Tuple2<Iterable<String>,Iterable<Integer>>> studentScores2 = studentsRDD.cogroup(scoresRDD);
studentScores2.foreach(new VoidFunction<Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>>>() {
   @Override
   public void call(Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> stu) throws Exception {
      System.out.println("stu id:"+stu._1);//1 3
      System.out.println("stu name:"+stu._2._1);//[xufengnian] [laohuang]
      System.out.println("stu score:"+stu._2._2);//[100,101] [80,81,71]
      Iterable<Integer> integers = stu._2._2;

      for (Iterator iter = integers.iterator(); iter.hasNext();) {
         Integer str = (Integer)iter.next();
         System.out.println(str);//100 101 80 81 71
      }
      System.out.println("===================================");
   }
});

到此，關(guān)于“如何使用spark Context轉(zhuǎn)成RDD”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！