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