package com.latrobe.spark

import org.apache.spark.{SparkContext, SparkConf}

/**
 * Created by spark on 15-1-19.
 * 根据key对K-V类型的RDD进行排序获得新的RDD
 */
object SortByKey {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    import org.apache.spark.SparkContext._
    val a = sc.parallelize(List("dog","cat","owl","gnu","ant"))
    val b = sc.parallelize(1 to a.count().toInt)
    val c = a.zip(b)

    //asc
    c.sortByKey(true).collect().foreach(print)
    //desc
    c.sortByKey(false).collect().foreach(print)
  }
}
/**
 * Created by spark on 15-1-19.
 * RDD1.subtract(RDD2):返回一个新的RDD，内容是：RDD1中存在的，RDD2中不存在的
 */
object Subtract {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 10)
    val b = sc.parallelize(1 to 3)
    //45678910
    //a.subtract(b).collect().foreach(print)

    val c = sc.parallelize(1 to 10)
    val d = sc.parallelize(List(1,2,3,11))
    //45678910
    c.subtract(d).collect().foreach(print)
  }
}
/**
 * Created by spark on 15-1-19.
 * RDD1.subtractByKey(RDD2):返回一个新的RDD，内容是：RDD1 key中存在的，RDD2 key中不存在的
 */
object SubtractByKey {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    import org.apache.spark.SparkContext._
    val a = sc.parallelize(List("dog","he","word","hello"))
    val b = a.keyBy(_.length)

    val c = sc.parallelize(List("cat","first","everyone"))
    val d = c.keyBy(_.length)
    //(2,he)(4,word)
    b.subtractByKey(d).collect().foreach(print)
  }
}
/**
 * Created by spark on 15-1-19.
 * sumApprox没有出现我希望的结果
 */
object SumAndSumApprox {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 1000000)
    val b = a.sum()
    val c = a.sumApprox(0L,0.9).getFinalValue()

    println(b + " *** " + c)
  }
}
/**
 * Created by spark on 15-1-19.
 * 取出RDD的前n个元素，以数组的形式返回
 */
object Take {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    //import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 1000000)

    //12345678910
    a.take(10).foreach(print)
  }
}
/**
 * Created by spark on 15-1-19.
 * 对RDD元素进行升序排序
 * 取出前n个元素并以数组的形式放回
 */
object TakeOrdered {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    //import org.apache.spark.SparkContext._
    val a = sc.parallelize(List("ff","aa","dd","cc"))

    //aacc
    a.takeOrdered(2).foreach(print)
  }
}
/**
 * Created by spark on 15-1-19.
 * 数据取样
 */
object TakeSample {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    //import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 10000)

    /**
     * 9048
        5358
        5216
        7301
        6303
        6179
        6151
        5304
        8115
        3869
     */
    a.takeSample(true , 10 , 1).foreach(println)
  }
}
/**
 * Created by spark on 15-1-19.
 * debug 详情信息显示
 */
object ToDebugString {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    //import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 9)
    val b = sc.parallelize(1 to 3)
    val c = a.subtract(b)

    c.toDebugString
  }
}
/**
 * Created by spark on 15-1-19.
 * 获得前几个最大值
 */
object Top {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    //import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 1000)
    val c = a.top(10)

    /**
     *1000
      999
      998
      997
      996
      995
      994
      993
      992
      991
     */
    c.foreach(println)
  }
}

/**
 * Union == ++ 把两个RDD合并为一个新的RDD
 */
object Union {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")
    val sc = new SparkContext(conf)

    //import org.apache.spark.SparkContext._
    val a = sc.parallelize(1 to 3)
    val b = sc.parallelize(3 to 5)
    val c = a.union(b)
    val d = a ++ b

    /**
     *123345
     */
    c.collect().foreach(print)

    /**
     *123345
     */
    d.collect().foreach(print)
  }
}

spark sortByKey subtractByKey take takeOrdered等函数使用例子