Spark算子学习(一)

原创
2017/03/19 20:43
阅读数 176

    2017.3.19。开始静下心来学习Spark,环境还没搭建好,但不妨碍我玩本地呀,是不是!

    工具:

    IDEA没了。

    最近喜欢上了画脑图,所以最后上脑图。

    1.map算子

        一一对应型,比如我将一个List传入map,map中可以通过匿名函数将这个list里的各项按照函数转换成一个新的RDD

    

    

def main(args: Array[String]): Unit = {
  //设定spark的配置信息。这里设定为在本地运行
  var conf = new SparkConf().setAppName("map算子").setMaster("local")
  //建立spark的上下文实例
  var sc = new SparkContext(conf)
  //新建一个数组
  val cao : Array[String] = Array("Hello","World")
  //将这个数组变成RDD,通过RDD的map算子将cao里面的每个值变成了 ***** 1
  sc.parallelize(cao).map(v => v + "\t1").foreach(i => println(i))

    2.filter算子

        对元素进行过滤。元素都要经过filter中定义的函数体来进行过滤,true留在RDD中

        

        

def main(args: Array[String]): Unit = {
  //啥都不想先建立spark链接
  var conf = new SparkConf().setAppName("我靠,我还是自己写名字吧").setMaster("local")
  var sc = new SparkContext(conf)
  var arr:Array[Int] = Array(1,2,3,4,5)
 sc.parallelize(arr).filter(v => v>2).foreach(i => println(i))

    3.faltMap算子

        同Map一样,但是map是一个元素对应一个输出元素,而flatMap则是可能会产出0个或多个成员

        

        

var conf = new SparkConf().setAppName("名字瞎写了").setMaster("local")
var sc = new SparkContext(conf)

var list:List[String] = List("Hello,World","HelloScala","Hello,Spark")
var arr2 = sc.parallelize(list).flatMap(f => f.split(",")).foreach(m => println(m))
var list1:List[Array[String]] = List(Array("A","B","C","D"),Array("E","F","G"))
var res = sc.parallelize(list1).flatMap(v => {

  for(i <- 0 until v.length){
   v(i) = v(i)+"\t1"
  }
 v
}).foreach(f => println(f))
结果是:
A 1
B 1
C 1
D 1
E 1
F 1
G 1

    4.mapPartitions算子
          同map类似,但是map针对的是RDD中的每个元素,而mapPartitions却是对RDD的一个分区进行参数是一个迭代器

            

            

每个分区得到一个迭代器,对迭代器操作就可以了。其实我个人觉得和map算子一样吧。下面代码
var conf = new SparkConf().setAppName("真不知道怎么取名字").setMaster("local")
var sc = new SparkContext(conf)

var list: List[String] = List("Hello", "Scala", "Hello", "Spark")
sc.parallelize(list).mapPartitions(f => {
  var arr = ArrayBuffer[String]()
  while(f.hasNext){
    arr.append(f.next()+"\t1")
  }
  arr.iterator
}).foreach(f => println(f))
而且感觉更加复杂了有没有啊。
 

    5.mapPartitionsWithIndex算子
         和mapPartitions差不多,只是多了一个参数,这个参数就是索引值

          

           

var conf = new SparkConf().setAppName("还是不知道怎么取名字").setMaster("local")
var sc = new SparkContext(conf)
var list = List("xx1","xx2","xx3","xx4","xx5","xx6","xx7","xx8","xx9")
sc.parallelize(list,3).mapPartitionsWithIndex((m,n) =>{
  var arr = ArrayBuffer[String]()
  while (n.hasNext){
   arr.append("分区索引为:"+m+"\t"+n.next)
 }
 arr.iterator
}).foreach(f => println(f))

以上代码解释解释,sc.parallelize(list,3)代表3个元素分成一个组,从0开始。之后跟的算子,参数就是两个所以定义m,n两个参数,m给索引,n为迭代器
结果是:
分区
索引为:0 xx1
分区索引为:0 xx2
分区索引为:0 xx3

分区索引为:1 xx4
分区索引为:1 xx5
分区索引为:1 xx6

分区索引为:2 xx7
分区索引为:2 xx8
分区索引为:2 xx9

    6.sample算子
     抽样算子,三个参数,第一个参数代表抽样元素是否返回重新抽样,true为返回,第二个为抽样比例,第三个随机数种子,这里就不举例了

    7.union算子
      合并两个RDD元素,元素类型必须一致,不去重

        

var conf = new SparkConf().setAppName("XX在看我,所以我要装逼")setMaster("local")
var sc = new SparkContext(conf)
var list1 = List("xx1")
var list2 = List("李x2")
//sc.parallelize(list1).union(sc.parallelize(list2)).foreach(f => println(f))
(sc.parallelize(list1)++sc.parallelize(list2)).foreach(f => println(f))

    8.intersection算子
      返回两个RDD的相同值,也就是说两个RDD的交集。

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部