Spark KV类型算子案例详解一

2021-10-20 23:59:53 阅读：404 来源： 互联网

标签：String val rdd Int RDD KV 算子 new Spark

1.
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

//kv类型算子
//PartitionBy：指定分区器来对上游的数据进行分区
object _01PartitonByDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("getRDD")
val sc = new SparkContext(conf)

//注意 RDD必须是对偶类型才会调出PartitonBy方法
//因为RDD隐式装换成立PairRDD
val rdd: RDD[(String, Int)] = sc.makeRDD(List(("1", 2), ("2", 3), ("4", 4), ("3", 0)), 2)
//自己指定分区器 3个生成3个分区
val result: RDD[(String, Int)] = rdd.partitionBy(new HashPartitioner(3))
result.collect().foreach(println)
//可以生成文件查看数据所在分区
result.saveAsTextFile("_01")
}
}

//(3,0)
//(1,2)
//(4,4)
//(2,3)

2.
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
//按照key分组，组内进行规约运算
object _02ReduceBykeyDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("getRDD")
val sc = new SparkContext(conf)

val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
//按照指定参数传入一个函数
val result: RDD[(String, Int)] = rdd.reduceByKey((x: Int, y: Int) => {
x + y
})
//可以查看分区数
val result1: RDD[(Int, (String, Int))] = result.mapPartitionsWithIndex((index, iter) => {
iter.map((index, _))
})
result1.collect().foreach(println)

//0,(b,4))
//(1,(a,5))
//(1,(c,0))
}
}

3.
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//作用：通过key进行分组操作，默认使用Hashpartitioner
//相同key的数据分到一个组中，形成一个对偶元组，
//   元组中的第一个元素，就是Key
//   元组中的第二个元素，就是同一个Key的多个value对应的迭代器
object _03GroupBykeyDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("getRDD")
val sc = new SparkContext(conf)

val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
//可以自己指定分区数
val result: RDD[(String, Iterable[Int])] = rdd.groupByKey(4)
result.collect().foreach(println)
println("*************************")

//需求2：按照要处理的RDD的元素的第二个值进行自定义分组
val result2: RDD[(Int, Iterable[(String, Int)])] = rdd.groupBy((e: (String, Int)) => e._2)
result2.saveAsTextFile("_03")

//(b,CompactBuffer(4))
//(c,CompactBuffer(0))
//*************************
//还有相应的生成三个分区

}
}

4.
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

// 第一个参数列表
// * 默认值
// * 第二个参数列表：
// * seqOp: 表示预聚合的函数，也就是分区内的运算逻辑
// * combOp：表示分区间的运算逻辑
object _04AggregateDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("getRDD")
val sc = new SparkContext(conf)

val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
/**
* 逻辑就是先在分区内逻辑运算，在再分区间逻辑运算
*/
val result: RDD[(String, Int)] = rdd.aggregateByKey(0)(
(x, y) => math.max(x, y)
,
(x, y) => x + y
)
result.collect().foreach(println)
//(b,4)
//(a,3)
//(c,0)
}
}

5.
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//aggregate函数的特例
//就是分区内和分区间的逻辑一样
object _05FloDByKeyDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("getRDD")
val sc = new SparkContext(conf)

val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
//使用一样的逻辑
val result: RDD[(String, Int)] = rdd.foldByKey(0)(
(x, y) => x + y
)
result.collect().foreach(println)

//(b,4)
//(a,5)
//(c,0)
}
}

标签：String,val,rdd,Int,RDD,KV,算子,new,Spark
来源： https://blog.csdn.net/Dream_aju/article/details/120877662

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Spark KV类型算子案例详解一