Spark算子——转换操作

概述

每一次转换操作都会产生不同的RDD,供给下一个操作使用。

算子

解决问题其实是将问题的初始化状态，通过一系列的操作Operate对问题的状态进行转换，然后达到完成解决的状态

惰性机制

RDD的转换过程是惰性求值的，也就是，整个转换过程只记录轨迹，并不会发生真正的计算，只有遇到了行动操作时，才会触发真正的计算。

filter(func)

过滤出满足函数func的元素，并返回存入一个新的数据集

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(List(1,2,3,4,5,6))
    val result = rdd.filter(_%2==0)
    println(result.collect().mkString(","))

map(func)

将每个元素传递到函数func中进行操作，并将结果返回为一个新的数据集。
collect()以数组的形式返回rdd的结果，但列表中每个数乘以2

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(List(1,2,3,4,5,6))
    val mapResult = rdd.map(_*2)
    println(mapResult.collect().toBuffer)

flatMap(func)

与map相似，但是每个输入元素都可以映射到0或多个输出结果,所以func应该返回一个序列，而不是单一元素

    val conf = new SparkConf().setAppName("RDD").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val arrayRDD: RDD[List[Int]] = sc.makeRDD(Array(List(1,2),List(3,4)))
    val listRDD: RDD[Int] = arrayRDD.flatMap(data=>data)
    listRDD.collect().foreach(println)

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(Array("a b c","b c d"))
    val result = rdd.flatMap(_.split(" "))
    println(result.collect().mkString(","))

sample

参数1 是否抽出的数据放回
参数2 抽样比例 浮点型
参数3 种子，默认值

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(1 to 10)
    val result = rdd.sample(false,0.5)
    println(result.collect().mkString(","))

union

求并集

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd1 = sc.parallelize(List(1,3,4))
    val rdd2 = sc.parallelize(List(2,3,4))
    val result = rdd1.union(rdd2)
    println(result.collect().toBuffer)

intersection

求交集

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd1 = sc.parallelize(List(1,3,4))
    val rdd2 = sc.parallelize(List(2,3,4))
    val result = rdd1.intersection(rdd2)
    println(result.collect().toBuffer)

distinct

去除重复元素

    val conf = new SparkConf().setAppName("spark").setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(List(1,3,4,3,5,1))
    val result = rdd.distinct()
    println(result.collect().toBuffer)

groupByKey(func)

应用于(K,V)键值的数据集时，返回一个新的（K,Iterable）形式的数据集

reduceByKey(func)

应用于(K,V)键值对的数据集时，返回一个新的(K,V)形式的数据集，其中每个值是将每个Key传递到函数func中进行聚合后的结果。

scala> val list = List("Hadoop","Spark","Hive","Scala")
list: List[String] = List(Hadoop, Spark, Hive, Scala)

scala> val rdd = sc.parallelize(list)
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[36] at parallelize at <console>:26

scala> val mapRDD = rdd.map(word=>(word,1))
mapRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[37] at map at <console>:25

scala> mapRDD.foreach(println)
(Hadoop,1)
(Spark,1)
(Hive,1)
(Scala,1)