摘要：摘要是一個(gè)項(xiàng)目，它被標(biāo)榜為快如閃電的集群計(jì)算。它擁有一個(gè)繁榮的開(kāi)源社區(qū)，并且是目前最活躍的項(xiàng)目。提供了一個(gè)更快更通用的數(shù)據(jù)處理平臺(tái)。更多經(jīng)典案例介紹期待下回分解。。。

Spark是一個(gè)Apache項(xiàng)目，它被標(biāo)榜為“快如閃電的集群計(jì)算”。它擁有一個(gè)繁榮的開(kāi)源社區(qū)，并且是目前最活躍的Apache項(xiàng)目。Spark提供了一個(gè)更快、更通用的數(shù)據(jù)處理平臺(tái)。和Hadoop相比，Spark可以讓你的程序在內(nèi)存中運(yùn)行時(shí)速度提升100倍，或者在磁盤上運(yùn)行時(shí)速度提升10倍。同時(shí)spark也讓傳統(tǒng)的map reduce job開(kāi)發(fā)變得更加簡(jiǎn)單快捷。本文將簡(jiǎn)單介紹幾個(gè)經(jīng)典hadoop的mr按理用spark實(shí)現(xiàn)，來(lái)讓大家熟悉spark的開(kāi)發(fā)。

求最大值最小值一直是Hadoop的經(jīng)典案例，我們用Spark來(lái)實(shí)現(xiàn)一下，借此感受一下spark中mr的思想和實(shí)現(xiàn)方式。話不多說(shuō)直接上code：

@Test
  def testMaxMin: Unit = {
    val sconf = new SparkConf().setAppName("test")
    val sc = new SparkContext(sconf)
    //初始化測(cè)試數(shù)據(jù)
    val data = sc.parallelize(Array(10,7,3,4,5,6,7,8,1001,6,2))
    //方法一
    val res = data.map(x => ("key", x)).groupByKey().map(x => {
      var min = Integer.MAX_VALUE
      var max = Integer.MIN_VALUE
      for(num <- x._2){
        if(num>max){
          max = num
        }
        if(num {
      println("max	"+x._1)
      println("min	"+x._2)
    })

    //方法二，下面用一個(gè)比較雞賊的方式求最大最小值
    val max = data.reduce((a,b) => Math.max(a,b))
    val min = data.reduce((a,b) => Math.min(a,b))
    println("max : " + max)
    println("min : " + min)
    sc.stop
  }

預(yù)期結(jié)果：

max: 1001
min: 2

思路和hadoop中的mr類似，設(shè)定一個(gè)key，value為需要求最大與最小值的集合，然后再groupBykey聚合在一起處理。第二個(gè)方法就更簡(jiǎn)單，性能也更好。

求每個(gè)key對(duì)應(yīng)的平均值是常見(jiàn)的案例，在spark中處理類似問(wèn)題常常會(huì)用到combineByKey這個(gè)函數(shù)，詳細(xì)介紹請(qǐng)google一下用法，下面看代碼：

@Test
  def testAvg(): Unit ={
    val sconf = new SparkConf().setAppName("test")
    val sc = new SparkContext(sconf)
    //初始化測(cè)試數(shù)據(jù)
    val foo = sc.parallelize(List(Tuple2("a", 1), Tuple2("a", 3), Tuple2("b", 2), Tuple2("b", 8)));
    //這里需要用到combineByKey這個(gè)函數(shù)，需要了解的請(qǐng)google
    val results=foo.combineByKey(
      (v)=>(v,1),
      (acc:(Int,Int),v) =>(acc._1+v,acc._2+1),
      (acc1:(Int,Int),acc2:(Int,Int))=>(acc1._1+acc2._1,acc1._2+acc2._2)
    ).map{case(key,value)=>(key,value._1/value._2.toDouble)}
    results.collect().foreach(println)
  }

我們讓每個(gè)partiton先求出單個(gè)partition內(nèi)各個(gè)key對(duì)應(yīng)的所有整數(shù)的和 sum以及個(gè)數(shù) count，然后返回一個(gè)pair(sum, count)在shuffle后累加各個(gè)key對(duì)應(yīng)的所有sum和count,再相除得到均值.

Top n問(wèn)題同樣也是hadoop種體現(xiàn)mr思想的經(jīng)典案例,那么在spark中如何方便快捷的解決呢：

@Test
  def testTopN(): Unit ={
    val sconf = new SparkConf().setAppName("test")
    val sc = new SparkContext(sconf)
    //初始話測(cè)試數(shù)據(jù)
    val foo = sc.parallelize(Array(
      ("a", 1),
      ("a", 2),
      ("a", 3),
      ("b", 3),
      ("b", 1),
      ("a", 4),
      ("b", 4),
      ("b", 2)
    ))
    //這里測(cè)試，取top 2。
    val groupsSort=foo.groupByKey().map(tu=>{
      val key=tu._1
      val values=tu._2
      val sortValues=values.toList.sortWith(_>_).take(2)
      (key,sortValues)
    })
    //轉(zhuǎn)換格式進(jìn)行print
    val flattenedTopNPerGroup =
      groupsSort.flatMap({case (key, numbers) => numbers.map(key -> _)})
    flattenedTopNPerGroup.foreach((value: Any) => {
      println(value)
    })
    sc.stop

  }

思路很簡(jiǎn)單，把數(shù)據(jù)groupBykey以后按key形成分組然后取每個(gè)分組最大的2個(gè)。預(yù)期結(jié)果：

(a,4)
(a,3)
(b,4)
(b,3)

以上簡(jiǎn)單介紹了一下hadoop中常見(jiàn)的3個(gè)案例在spark中的實(shí)現(xiàn)。如果讀者們已經(jīng)接觸過(guò)或者寫過(guò)一些hadoop的mapreduce job，那么會(huì)不會(huì)覺(jué)得在spark中寫起來(lái)方便快捷很多呢。

更多spark經(jīng)典案例介紹期待下回分解。。。

作者信息
MaxLeap團(tuán)隊(duì)_數(shù)據(jù)分析組 成員：譚楊【原創(chuàng)】
首發(fā)自：https://blog.maxleap.cn/archi...

作者往期佳作
淺析時(shí)間序列數(shù)據(jù)
淺析Apache Spark Caching和Checkpointing