Spark常用算子解析及代码实战

Spark

一.coalesce

　　1.简介

　　　　coalesce常用来合并分区，第二个参数是合并分区时是否产生shuffle。true为产生shuffle，false为不产生shuffle。默认是false不产生shuffle。如果coalesce设置的分区数比原来的分区数还大的话若设置为false则不起作用。如果设置为true则效果等价于repartition。即repartition(numPartitions) = coalesce(numPartitions)。

　　2.测试数据

　　　　val array = Array("spark,scala,6", "hadoop,java,12", "tensorflow,python,8", "solr,java,16", "hbase,java,11")

　　3.代码　　　　　　　　　　　

　　　　/**
  　　　　* coalesce算子，常用于减少分区
 　　　　 */
　　　　val befParNum = rdd.getNumPartitions
　　　　rdd = rdd.coalesce(1, false) // true为产生shuffle
　　　　val coalParNum = rdd.getNumPartitions
　　　　/**
 　　　　 * repartition与之类型，一般增大分区数
 　　　　 */
　　　　rdd = rdd.repartition(3)
　　　　val reParNum = rdd.getNumPartitions
　　　　println("初始分区数：" + befParNum + ",coalesce分区后：" + coalParNum + ",repartition分区后：" + reParNum)

　　4.结果

　　　　初始分区数：2,coalesce分区后：1,repartition分区后：3

二.zip，zipWithIndex

　　1.简介

　　　　zip将两个RDD中的元素变成一个KV格式的RDD，两个RDD的每个分区元素个数必须相同。zipWithIndex该函数将RDD中的元素和这个元素在RDD中的索引下标【从0开始】组合成【K,V】键值对。

　　2.测试数据

　　 val zip_array_left = Array(1,2,5,6,7,5,3,1)
　　 val zip_array_left_2 = Array(1,2,5,6,7,8,9,0)
　　 val zip_array_right = Array("spark", "scala", "hive", "hbase", "python", "hive", "hbase", "hbase")

　　3.代码

　　 /**
  　 　* zip
 　 　 */
　　 //to rdd
　 　val zip_left = sc.parallelize(zip_array_left)
　 　val zip_left_2 = sc.parallelize(zip_array_left_2)
　 　val zip_right = sc.parallelize(zip_array_right)

　 　//zip
　 　val zip = zip_left.zip(zip_right)
　 　zip.foreach(println)
　 　println("------------------")

　 　val zip_2 = zip_left_2.zip(zip_right)
　 　/**
　 　  * zipWithIndex
   　　*/
　 　val zip_index = zip.zipWithIndex()
　 　zip_index.foreach(println)

　　4.结果

　　　　(1,spark)
　　　　(2,scala)
　　　　(5,hive)
　　　　(6,hbase)
　　　　(7,python)
　　　　(5,hive)
　　　　(3,hbase)
　　　　(1,hbase)
　　　　------------------
　　　　((1,spark),0)
　　　　((7,python),4)
　　　　((2,scala),1)
　　　　((5,hive),2)
　　　　((5,hive),5)
　　　　((6,hbase),3)
　　　　((3,hbase),6)
　　　　((1,hbase),7)

三.countByKey，countByValue

　　1.简介

　　　　countByKey作用在K,V格式的RDD之上，统计相同key的个数。countByValue作用在K,V格式的RDD之上，统计相同value的个数。

　　2.测试数据

　　　　同上

　　3.代码

　 　/**
 　　  * countByKey
　　   */
　 　val zip_key = zip.countByKey()
　　 zip_key.foreach(println)
　　 println("------------------")
　　 /**
　　   * countByValue
　　   */
　　 val zip_value = zip.countByValue()
　　 zip_value.foreach(println)

　　4.结果　

　　　　(5,2)
　　　　(1,2)
　　　　(6,1)
　　　　(2,1)
　　　　(7,1)
　　　　(3,1)
　　　　------------------
　　　　((7,python),1)
　　　　((1,spark),1)
　　　　((2,scala),1)
　　　　((1,hbase),1)
　　　　((3,hbase),1)
　　　　((6,hbase),1)
　　　　((5,hive),2)

四.cogroup

　　1.简介

　　　　cogroup 对两个内部数据结构为元组（仅有两个元素的元组）的数据进行匹配，把匹配上的value值保存到一个元组中。

　　2.测试数据

　　　　同上

　　3.代码　　

　　 zip.cogroup(zip_2).foreach(println)

　　4.结果

　　　　(0,(CompactBuffer(),CompactBuffer(hbase)))
　　　　(1,(CompactBuffer(spark, hbase),CompactBuffer(spark)))
　　　　(7,(CompactBuffer(python),CompactBuffer(python)))
　　　　(3,(CompactBuffer(hbase),CompactBuffer()))
　　　　(6,(CompactBuffer(hbase),CompactBuffer(hbase)))
　　　　(9,(CompactBuffer(),CompactBuffer(hbase)))
　　　　(8,(CompactBuffer(),CompactBuffer(hive)))
　　　　(5,(CompactBuffer(hive, hive),CompactBuffer(hive)))
　　　　(2,(CompactBuffer(scala),CompactBuffer(scala)))　

五.flatten

　　1.简介

　　　　把多层集合数据展开成一个集合。

　　2.测试数据

　　 val sourceDate = Array("zhen@zhen01/2018-09-04_18;57;02_SOURCE",
 　　  "zhen@zhen02/2018-09-05_11;37;11_SOURCE","zhen@zhen03/2018-09-06_11;37;11_TEST")
　　 val resultDate = Array("zhen@zhen01/2018-09-04_18;57","zhen@zhen02/2018-09-05_11;37",
 　 　 "zhen@zhen03/2018-09-06_11;37")

　　3.代码

　　 val seq = Seq(sourceDate, resultDate)
　　 seq.flatten.foreach(println)

　　4.结果

　　　　zhen@zhen01/2018-09-04_18;57;02_SOURCE
　　　　zhen@zhen02/2018-09-05_11;37;11_SOURCE
　　　　zhen@zhen03/2018-09-06_11;37;11_TEST
　　　　zhen@zhen01/2018-09-04_18;57
　　　　zhen@zhen02/2018-09-05_11;37
　　　　zhen@zhen03/2018-09-06_11;37