ApacheSparkBook 第三章《Spark 逻辑处理流程》勘误与修改建议

Jul 20 '20 06:07 JerryLead

Page 65, 图 3.13 的 2,k, 3,e 为啥不是2,k1, 3,e1

Aug 30 '20 05:08 isplendid

@isplendid 因为combineByKey中的createCombiner()只会作用于相同key的第一个record。在本例中，处理<2, b>时发现key=2没有被处理过，所以使用createCombiner()将b转换为b1，得到<2, b1>，保存在内存中。接下来处理<2, k>，由于key=2已经被处理过（在内存中），所以使用mergeValue()来处理<2, k>，也就是mergeValue(<2, b1>, <2, k>) => <2, b1+k>。

可以使用下面的示例程序来验证：


import org.apache.spark.sql.SparkSession

object CombineByKeyExample {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .appName("combine by key example")
      .master("local[2]")
      .getOrCreate()

    val sc = spark.sparkContext
    val inputRDD = sc.parallelize(Array[(Int, Char)](
      (1, 'a'), (2, 'b'), (2, 'k'), (3, 'c'), (4, 'd'), (3, 'e'),
      (3, 'f'), (2, 'g'), (2, 'h')
    ), 3)

    println("-----------input RDD--------")
    inputRDD.mapPartitionsWithIndex((pid, iter) => {
      iter.map(value => "PID: " + pid + ", value: " + value)
    }).foreach(println)

    val resultRDD = inputRDD.combineByKey((v: Char) => {
      if (v == 'c') {
        v + "0"
      } else {
        v + "1"
      }
    }
      , (c: String, v: Char) => c + "+" + v, (c1: String, c2: String) => c1 + "_" + c2, 2)
    //    val resultRDD = inputRDD.combineByKey((v:Char)=>List(v), (c:List[Char],v:Char)=>v::c,(c1:List[Char],c2:List[Char])=>c1:::c2)
    println(resultRDD.toDebugString)
    println("-----------result RDD--------")
    resultRDD.mapPartitionsWithIndex((pid, iter) => {
      iter.map(value => "PID: " + pid + ", value: " + value)
    }).foreach(println)
  }
}

Aug 30 '20 13:08 JerryLead

Page 61, aggregateByKey()操作的标题前的倒数第三行末尾起在性能上，相比groupBykey()、reduceByKye()可以在Shuffle之前使用func对数据进行聚合，减少了…… 这边的顿号应该改为逗号