《深入理解Spark》之運算元詳解

阿新 • • 發佈：2019-01-20

XML Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

package com.lyzx.day17

import org.apache.spark.{SparkContext,SparkConf}

class T1 {
  /*
    cartesian  做笛卡爾積
    RDD[K]+RDD[V]=>RDD[(K,V)]
    把每項依次組合後放入一個元組中返回
   */
  def f1(sc:SparkContext): Unit ={
    val arr1 = List("A","B","C")
    val arr2 = List(1,2,3)

    val rdd1 = sc.parallelize(arr1)
    val rdd2 = sc.parallelize(arr2)

    rdd1.persist()
    rdd2.persist()

    rdd1.cartesian(rdd2).foreach(println)

    println("==============================="

)

    val rdd11 = rdd1.map(item=>(item,item))
    val rdd21 = rdd2.map(item=>(item,item))
    rdd11.cartesian(rdd21).foreach(println)
  }

  /*
  countByKey  是一個action操作  計算每一個key出現的次數
  具有shuffle操作
  計算出每一個鍵對應的個數,底層使用reduceByKey
   */
  def f2(sc:SparkContext): Unit ={
    val arr1 = (1 to 10)
    val rdd1 = sc.parallelize(arr1)

    val arr2 = (5 to 15)
    val rdd2 = sc.parallelize(arr2)

    rdd1.map(item=>

(item,item*100))
        .union(rdd2.map(item=>(item,item+99)))
        .countByKey()
        .foreach(println)
  }

  /*
  join
    類似於SQL裡面的inner join左表根據條件匹配右表的記錄,如果左表沒有匹配到右表的記錄則不會加回去
    返回元組,類似於(key,(value1,value2,...,values3))
   cogroup
    首先在左表的記錄(k,v)封裝為(k,compactBuffer(v))的形式再做join操作把匹配的項(k,v1),(k,v2)封裝為(k,compactBuffer(v1,v2))
    右表做相同的操作,然後左表右表做join操作
   */
  def f3(sc:SparkContext): Unit ={
    val arr1 = List((1,"羅永浩"),(2,"雷軍"),(3,"餘承東"),(1,"玉皇大帝"),(4,"戰神刑天"))
    val studentRdd = sc.parallelize(arr1)
    studentRdd.persist()

    val arr2 = List((1,88),(2,99),(3,99))
    val scoreRdd = sc.parallelize(arr2)
    scoreRdd.persist()

    //studentRdd中的記錄根據key匹配scoreRdd中的記錄,然後把匹配的記錄放入到一個元組中(key,(leftValue,rightValue))
    studentRdd.join(scoreRdd).foreach(println)

    println("= = = = = = = = = = = = = = = = = = = = = ")

    //cogroup 首先在studentRdd中把鍵歸併在一起把值存入CompactBuffer  在scoreRdd中相同
    //然後把兩個計算的結果在做join
    studentRdd.cogroup(scoreRdd).foreach(println)
  }

  /*
    mapValues 只針對鍵值對的RDD，其中鍵不變,後面的函式只操作值
   */
  def f4(sc:SparkContext): Unit ={
    val rdd = sc.parallelize((1 to 10))
    val mapRdd = rdd.map(item=>(item,item+100))
    mapRdd.mapValues(_+1).foreach(println)
  }

  /*
    有序列化的地方
    1、儲存等級  MEMERY_ONLY > MEMERY_SER
    2、shuffle 走網路傳輸的時候
    3、分發任務和返回結果的時候

    讀取本地檔案
    textFile()的第二個引數是最少的partition的個數
    如果不傳就找叢集中的預設的並行度和2的最小值最為預設的partition的個數
    如果讀取hdfs上的檔案預設有多少個block就有多少個partition
    如果出入的引數小於block的個數則partition的個數還是block個
    如果大於block的個數則就使用引數個
   */

  /*
    sortBy
    第一個引數是一個函式，該函式的也有一個帶T泛型的引數，返回型別和RDD中元素的型別是一致的；
      >>主要用於對每個元素進行計算轉換之類的操作
      >>如果不需要做計算就寫x=>x  從x到x的對映
      >>後面兩個引數都有預設值，如果想降序排列可以一個引數搞定x=>{-x}
　　第二個引數是ascending，從字面的意思大家應該可以猜到，是的，這引數決定排序後RDD中的元素是升序還是降序，預設是true，也就是升序；
      >>是否是升序
　　第三個引數是numPartitions，該引數決定排序後的RDD的分割槽個數，預設排序後的分割槽個數和排序之前的個數相等，
    即為this.partitions.size
   */
  def f5(sc:SparkContext): Unit ={
    val rdd = sc.parallelize(List(10,9,8,7,6,5,4,3,2,1))
    rdd.sortBy(x=>x,true).foreach(print)
    println("========================================")
    val rdd2 = sc.parallelize(1 to 10)
    rdd2.sortBy(x=>{-x}).foreach(print)
  }

  /*
    countByValue
    數一數值有多少個,這兒的value並不是鍵值對裡面的value
    而是RDD的裡面的元素，返回結果是一個元組(element,count) 第一項是元素，第二項是個數
   */
  def f6(sc:SparkContext): Unit ={
    val rdd = sc.parallelize(List((1,10),(2,10),(3,3),(4,4),(5,10),(5,10),(6,1),(7,1)))
//    val mapRdd = rdd.map(item=>(item,item))
    val xx = rdd.countByValue()
    xx.foreach(println)

    println("==================================")
    val rdd2 = sc.parallelize(List(1,2,2,2,3,4))
    val result = rdd2.countByValue()
    result.foreach(item=>println(item._1+"-"+item._2))
  }
  /*
     groupBy  分組組成鍵值對的形式
     鍵是每元素  值是元素組成的CompactBuffer集合
   */
  def f7(sc:SparkContext): Unit ={
    val rdd = sc.parallelize(List(1,2,2,2,3,4,5,6))
    rdd.groupBy(x=>x).foreach(println)
  }

  /*
  一個簡單的TopN 小栗子
  如下data列表中每一個元素都是網名在搜尋引擎中搜索的關鍵字，現在要統計今天的熱搜詞(出現次數最多的次)Top2
   */
  def f8(sc:SparkContext): Unit ={
    val data = List("A","B","B","A","C","D","A","A","A","A","A","A","A","A","A","A","A","C","C","C","C","C")
    val rdd = sc.parallelize(data)

    //org.apache.spark.rdd.ShuffledRDD

    val mapRdd = rdd.countByValue()
                    .map(item=>(item._2,item._1))
    println(mapRdd.getClass.getName)
    mapRdd.foreach(item=>println(item._1+"----"+item._2))

//    rdd.countByValue()
//      .map(item=>(item._2,item._1))
//      .toSeq
//      .sortBy(x=>{-x._1})
//      .take(2)
//      .foreach(item=>println(item._2))
  }

  def f9(sc:SparkContext): Unit ={
//    val rdd = sc.parallelize((1 to 10))
//    val mapRdd = rdd.map(item=>(item,item))
//    mapRdd
//    .sortByKey()
//    println(mapRdd.getClass.getName)

    val data1 = ("A","B","C")
    val data2 = (1 to 10)
    println(data1.getClass.getName)
    println(data2.getClass.getName)
  }
}

object T1{
  def main(args: Array[String]) {
      val conf = new SparkConf().setAppName("day17").setMaster("local")
      val sc = new SparkContext(conf)

      val t = new T1
//    t.f1(sc)
//    t.f2(sc)
//    t.f3(sc)
//    t.f4(sc)
//    t.f5(sc)
//    t.f6(sc)
//    t.f7(sc)
//    t.f8(sc)
    t.f9(sc)
      sc.stop()
  }
}

《深入理解Spark》之運算元詳解

深入理解spark－rdd詳解

深入理解Android之Xposed詳解

《深入理解Spark》之運算元詳解

《深入理解Spark》之Spark常用運算元詳解(java版+spark1.6.1)

Spark常用運算元詳解彙總：實戰案例、Java版本、Scala版本

Spark常用運算元詳解

C++深入理解單例模式詳解

【深入理解CSS】BFC詳解

深入理解Spark之ListenerBus監聽器

安卓專案實戰之強大的網路請求框架okGo使用詳解（二）：深入理解Callback之自定義JsonCallback

零基礎入門大資料之spark中rdd部分運算元詳解

深入理解Tomcat系列之七詳解URL請求

深入理解盒子模型——CSS之BFC詳解

《深入理解Spark》之通過sample運算元找出導致資料傾斜的key

Spark運算元篇 --Spark運算元之aggregateByKey詳解

Spark函數詳解系列之RDD基本轉換

Spark算子之aggregateByKey詳解

spark運算元詳解

深入React技術棧之setState詳解

深入學習理解(9):java:AbstractQueuedSynchronizer詳解

《深入理解Spark》之運算元詳解

相關推薦