建站
咨詢
售后
建站咨詢
新聞中心
Spark RDD常用算子：Value類型
Spark之所以比Hadoop靈活和強大，其中一個原因是Spark內(nèi)置了許多有用的算子，也就是方法。通過對這些方法的組合，編程人員就可以寫出自己想要的功能。說白了spark編程就是對spark算子的使用，下面為大家詳細(xì)講解一下SparkValue類型的常用算子
創(chuàng)新互聯(lián)建站是一家專注于成都網(wǎng)站制作、成都網(wǎng)站設(shè)計、外貿(mào)營銷網(wǎng)站建設(shè)與策劃設(shè)計,云陽網(wǎng)站建設(shè)哪家好?創(chuàng)新互聯(lián)建站做網(wǎng)站,專注于網(wǎng)站建設(shè)10年,網(wǎng)設(shè)計領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:云陽等地區(qū)。云陽做網(wǎng)站價格咨詢:13518219792
map
函數(shù)說明：
map() 接收一個函數(shù)，該函數(shù)將RDD中的元素逐條進行映射轉(zhuǎn)換，可以是類型的轉(zhuǎn)換，也可以是值的轉(zhuǎn)換，將函數(shù)的返回結(jié)果作為結(jié)果RDD編程。
函數(shù)簽名：
def map[U: ClassTag](f: T => U): RDD[U]
案例演示
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
//算子 -map
val rdd = sc.makeRDD(List(1, 2, 3, 4),2)
val mapRdd1 = rdd.map(
_*2
)
mapRdd1.collect().foreach(println)
sc.stop()
運行結(jié)果
2
4
6
8
mapPartitons
函數(shù)說明：
將待處理的數(shù)據(jù)以分區(qū)為單位發(fā)送到待計算節(jié)點上進行處理，mapPartition是對RDD的每一個分區(qū)的迭代器進行操作，返回的是迭代器。這里的處理可以進行任意的處理。
函數(shù)簽名：
def mapPartitions[U: ClassTag](
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]
案例演示
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
//算子 -mapPartitons 計算每個分區(qū)的最大數(shù)
val rdd = sc.makeRDD(List(1, 34, 36,345,2435,2342,62,35, 4),4)
val mapParRdd = rdd.mapPartitions(
iter => {
List(iter.max).iterator
}
)
mapParRdd.foreach(println)
sc.stop()
}
運行結(jié)果：
62
2435
34
345
mapPartitonsWithIndex
函數(shù)說明：
將待處理的數(shù)據(jù)以分區(qū)為單位發(fā)送到計算節(jié)點上，這里的處理可以進行任意的處理，哪怕是過濾數(shù)據(jù)，在處理的同時可以獲取當(dāng)前分區(qū)的索引值。
函數(shù)簽名：
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]
案例演示：
-
將數(shù)據(jù)進行扁平化映射并且打印所在的分區(qū)數(shù)
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List("Hello Spark", "Hello Scala", "Word Count"),2)
val mapRDD = rdd.flatMap(_.split(" "))
val mpwiRdd = mapRDD.mapPartitionsWithIndex(
(index, datas) => {
datas.map(
num => {
(index, num)
}
)
}
)
mpwiRdd.collect().foreach(println)
}
運行結(jié)果：
(0,Hello)
(0,Spark)
(1,Hello)
(1,Scala)
(1,Word)
(1,Count)
-
將數(shù)據(jù)進行扁平化映射只打印所在第一分區(qū)的數(shù)據(jù)
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List("Hello Spark", "Hello Scala", "Word Count"),2)
val mapRDD = rdd.flatMap(_.split(" "))
val mpwiRdd = mapRDD.mapPartitionsWithIndex(
(index, datas) => {
if (index==0){
datas.map(
num => {
(index, num)
}
)
}else{
Nil.iterator
}
}
)
mpwiRdd.collect().foreach(println)
運行結(jié)果：
(0,Hello)
(0,Spark)
flatMap
函數(shù)說明：
將數(shù)據(jù)進行扁平化之后在做映射處理，所以算子也稱為扁平化映射
函數(shù)簽名：
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
案例演示：
將每個單詞進行扁平化映射
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
//算子 -map
val rdd = sc.makeRDD(List("Hello Scala","Hello Spark"), 2)
val FltRdd = rdd.flatMap(
_.split(" ")
)
FltRdd.foreach(println)
sc.stop()
}
運行結(jié)果：
Hello
Scala
Hello
Spark
glom
函數(shù)說明：
glom的作用就是將一個分區(qū)的數(shù)據(jù)合并到一個array中。
函數(shù)簽名：
def glom(): RDD[Array[T]]
案例演示：
-
將不同分區(qū)rdd的元素合并到一個分區(qū)
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9),2)
val glomRdd = rdd.glom()
glomRdd.collect().foreach(data=>println(data.mkString(",")))
sc.stop()
}
運行結(jié)果：
1,2,3,4
5,6,7,8,9
groupBy
函數(shù)說明：
將數(shù)據(jù)根據(jù)指定的規(guī)則進行分組，分區(qū)默認(rèn)不變，單數(shù)數(shù)據(jù)會被打亂，我們成這樣的操作為shuffer，
函數(shù)簽名：
def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]
案例演示：
-
按照奇偶數(shù)進行g(shù)roupby分區(qū)
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8,10),2)
val groupByRDD = rdd.groupBy(_ % 2 == 0)
groupByRDD.collect().foreach(println)
sc.stop()
}
運行結(jié)果：
(false,CompactBuffer(1, 3, 5, 7))
(true,CompactBuffer(2, 4, 6, 8, 10))
-
按照單詞的首字母進行分組
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List("Hello","Tom","Timi","Scala","Spark"))
val groupByRDD = rdd.groupBy(_.charAt(0))
groupByRDD.collect().foreach(println)
sc.stop()
}
運行結(jié)果：
(T,CompactBuffer(Tom, Timi))
(H,CompactBuffer(Hello))
(S,CompactBuffer(Scala, Spark))
filter
函數(shù)說明：
filter即過濾器的意思，所以filter算子的作用就是過濾的作用。filter將根據(jù)指定的規(guī)則進行篩選過濾，符合條件的數(shù)據(jù)保留，不符合的數(shù)據(jù)丟棄，當(dāng)數(shù)據(jù)進行篩選過濾之后，分區(qū)不變，但分區(qū)內(nèi)的數(shù)據(jù)可能不均衡，生產(chǎn)環(huán)境下，可能會出現(xiàn)數(shù)據(jù)傾斜。
函數(shù)簽名：
def filter(f: T => Boolean): RDD[T]
案例演示：
-
篩選出能被二整除的數(shù)字
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List(46,235,246,2346,3276,235,234,6234,6245,246,24,6246,235,26,265))
val filterRDD = rdd.filter(_ % 2 == 0)
filterRDD.collect().foreach(println)
sc.stop()
}
運行結(jié)果：
46
246
2346
3276
234
6234
246
24
6246
26
2.篩選單詞中包含H的
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
val sc = new SparkContext(conf)
val rdd = sc.makeRDD(List("Hello","Horber","Hbeer","ersfgH","Scala","Hadoop","Zookeeper"))
val filterRDD = rdd.filter(_.contains("H"))
filterRDD.collect().foreach(println)
sc.stop()
}
運行結(jié)果：
Hello
Horber
Hbeer
ersfgH
Hadoop
網(wǎng)站名稱：SparkValue類型的常用算子
分享網(wǎng)址：http://www.5511xx.com/article/cospdjo.html
