在工作中經常遇到需要合併RDD的情況，記錄下處理情況。join和cogroup運算元都能達到要求，按key合併，只是當rdd存在多個相同的key時候，最終的輸出結果不一樣。網上找到了處理情況，自己也測試了，程式碼如下：

object Test {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("test").
      master("local[2]").getOrCreate()

    val sc = spark.sparkContext
    sc.setLogLevel("WARN")

    /**
      * id      name
      * 1       zhangsan
      * 2       lisi
      * 3       wangwu
      */
    val idName = sc.parallelize(Array((1, "zhangsan"), (2, "lisi"), (3, "wangwu")))

    /**
      * id      age
      * 1       30
      * 2       29
      * 4       21
      */
    val idAge = sc.parallelize(Array((1, 30), (2, 29), (4, 21)))

    println("cogroup")

    /**
      * (1,(CompactBuffer(zhangsan),CompactBuffer(30)))
      * (2,(CompactBuffer(lisi),CompactBuffer(29)))
      * (3,(CompactBuffer(wangwu),CompactBuffer()))
      * (4,(CompactBuffer(),CompactBuffer(21)))
      */
    idName.cogroup(idAge).collect().foreach(println)

    println("join")
    // fullOuterJoin於cogroup的結果類似, 只是資料結構不一樣
    /**
      * (1,(Some(zhangsan),Some(30)))
      * (2,(Some(lisi),Some(29)))
      * (3,(Some(wangwu),None))
      * (4,(None,Some(21)))
      */
    idName.fullOuterJoin(idAge).collect().foreach(println)

    /**
      * id      score
      * 1       100
      * 2       90
      * 2       95
      */
    val idScore = sc.parallelize(Array((1, 100), (2, 90), (2, 95)))

    println("cogroup, 出現相同id時")

    /**
      * (1,(CompactBuffer(zhangsan),CompactBuffer(100)))
      * (2,(CompactBuffer(lisi),CompactBuffer(90, 95)))
      * (3,(CompactBuffer(wangwu),CompactBuffer()))
      */
    idName.cogroup(idScore).collect().foreach(println)

    println("join, 出現相同id時")

    /**
      * (1,(Some(zhangsan),Some(100)))
      * (2,(Some(lisi),Some(90)))
      * (2,(Some(lisi),Some(95)))
      * (3,(Some(wangwu),None))
      */
    idName.fullOuterJoin(idScore).collect().foreach(println)
  }


}