devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

实验2-spark编程_请根据给定的实验数据,通过编程来计算以下内容: (1)该系总共有多少学生; (2)该系

作者：IT小白 | 2024-06-02 15:31:05

赞

踩

请根据给定的实验数据,通过编程来计算以下内容: (1)该系总共有多少学生; (2)该系

实验目的

（1）通过实验掌握Spark的基本编程方法；

（2）熟悉RDD到DataFrame的转化方法；

（3）熟悉利用Spark管理来自不同数据源的数据。

实验内容

1．Spark基本操作

请参照给出的数据score.txt，该数据集包含了某大学计算机系的成绩，数据格式如下所示：

 

Tom,DataBase,80

 

Tom,Algorithm,50

 

Tom,DataStructure,60

 

Jim,DataBase,90

 

Jim,Algorithm,60

 

Jim,DataStructure,80

 

……

请根据给定的实验数据，在spark-shell中通过编程来计算以下内容：

代码实现：

  1. #注意，首先将score.txt文件上传到hdfs的根目录下，然后将hadoop01和9000替换为自己的就OK了
  2. val scores = sc.textFile("hdfs://hadoop01:9000/score.txt")
  3.  
  4. //人数
  5. val totalStudents = scores.map(line => line.split(",")(0)).distinct().count()
  6.  
  7.  
  8. // (2) 计算该系共开设了多少门课程
  9. val totalCourses = scores.map(line => line.split(",")(1)).distinct().count()
  10.  
  11. // (3) 计算 Tom 同学的总成绩平均分
  12. val tomScores = scores.filter(_.startsWith("Tom")).map(_.split(",")(2).toInt)
  13. val tomAvgScore = tomScores.sum() / tomScores.count()
  14.  
  15. // (4) 计算每名同学的选修课程门数
  16. val courseCounts = scores.map(line => (line.split(",")(0), 1)).reduceByKey(_ + _)
  17.  
  18. // (5) 计算该系 DataBase 课程共有多少人选修
  19. val dbStudents = scores.filter(_.split(",")(1) == "DataBase").map(_.split(",")(0)).distinct().count()
  20.  
  21. // 输出结果
  22. println(s"该系总共有 ${totalStudents} 位学生")
  23. println(s"该系共开设了 ${totalCourses} 门课程")
  24. println(s"Tom 同学的总成绩平均分为 ${tomAvgScore} 分")
  25. println("每位同学选修的课程门数如下：")
  26. courseCounts.collect().foreach { case (student, count) =>
  27.   println(s"${student} 选修了 ${count} 门课程")
  28. }
  29. println(s"该系 DataBase 课程共有 ${dbStudents} 人选修")

实验过程：

  1. ################################LOG########################################
  2. scala> val scores = sc.textFile("hdfs://hadoop01:9000/score.txt")
  3. 2024-03-20 15:32:06,949 INFO memory.MemoryStore: Block broadcast_1 stored as values in memory (estimated size 389.1 KiB, free 365.5 MiB)
  4. 2024-03-20 15:32:06,978 INFO memory.MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 40.2 KiB, free 365.5 MiB)
  5. 2024-03-20 15:32:06,980 INFO storage.BlockManagerInfo: Added broadcast_1_piece0 in memory on hadoop01:38796 (size: 40.2 KiB, free: 366.2 MiB)
  6. 2024-03-20 15:32:06,982 INFO spark.SparkContext: Created broadcast 1 from textFile at <console>:23
  7. scores: org.apache.spark.rdd.RDD[String] = hdfs://hadoop01:9000/score.txt MapPartitionsRDD[4] at textFile at <console>:23
  8.  
  9. scala> val totalStudents = scores.map(line => line.split(",")(0)).distinct().count()
  10. 2024-03-20 15:32:14,939 INFO mapred.FileInputFormat: Total input files to process : 1
  11. 2024-03-20 15:32:15,160 INFO spark.SparkContext: Starting job: count at <console>:23
  12. 2024-03-20 15:32:15,628 INFO scheduler.DAGScheduler: Registering RDD 6 (distinct at <console>:23) as input to shuffle 0
  13. ......
  14. 2024-03-20 15:33:15,396 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 7.0 (TID 14) (hadoop01, executor driver, partition 0, NODE_LOCAL, 7181 bytes)
  15. 2024-03-20 15:33:15,396 INFO scheduler.TaskSetManager: Starting task 1.0 in stage 7.0 (TID 15) (hadoop01, executor driver, partition 1, NODE_LOCAL, 7181 bytes)
  16. 2024-03-20 15:33:15,397 INFO executor.Executor: Running task 1.0 in stage 7.0 (TID 15)
  17. 2024-03-20 15:33:15,397 INFO executor.Executor: Running task 0.0 in stage 7.0 (TID 14)
  18. 2024-03-20 15:33:15,399 INFO storage.ShuffleBlockFetcherIterator: Getting 2 (580.0 B) non-empty blocks including 2 (580.0 B) local and 0 (0.0 B) host-local and 0 (0.0 B) push-merged-local and 0 (0.0 B) remote blocks
  19. 2024-03-20 15:33:15,399 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 0 ms
  20. 2024-03-20 15:33:15,399 INFO storage.ShuffleBlockFetcherIterator: Getting 2 (580.0 B) non-empty blocks including 2 (580.0 B) local and 0 (0.0 B) host-local and 0 (0.0 B) push-merged-local and 0 (0.0 B) remote blocks
  21. 2024-03-20 15:33:15,399 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 0 ms
  22. 2024-03-20 15:33:15,406 INFO executor.Executor: Finished task 0.0 in stage 7.0 (TID 14). 1747 bytes result sent to driver
  23. 2024-03-20 15:33:15,406 INFO executor.Executor: Finished task 1.0 in stage 7.0 (TID 15). 1747 bytes result sent to driver
  24. 2024-03-20 15:33:15,407 INFO scheduler.TaskSetManager: Finished task 1.0 in stage 7.0 (TID 15) in 11 ms on hadoop01 (executor driver) (1/2)
  25. 2024-03-20 15:33:15,407 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 7.0 (TID 14) in 11 ms on hadoop01 (executor driver) (2/2)
  26. 2024-03-20 15:33:15,407 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 7.0, whose tasks have all completed, from pool
  27. 2024-03-20 15:33:15,408 INFO scheduler.DAGScheduler: ResultStage 7 (count at <console>:23) finished in 0.019 s
  28. 2024-03-20 15:33:15,408 INFO scheduler.DAGScheduler: Job 4 is finished. Cancelling potential speculative or zombie tasks for this job
  29. 2024-03-20 15:33:15,408 INFO scheduler.TaskSchedulerImpl: Killing all running tasks in stage 7: Stage finished
  30. 2024-03-20 15:33:15,409 INFO scheduler.DAGScheduler: Job 4 finished: count at <console>:23, took 0.057629 s
  31. dbStudents: Long = 125
  32.  
  33. scala> println(s"该系总共有 ${totalStudents} 位学生")
  34. 该系总共有 265 位学生
  35.  
  36. scala> println(s"该系共开设了 ${totalCourses} 门课程")
  37. 该系共开设了 8 门课程
  38.  
  39. scala> println(s"Tom 同学的总成绩平均分为 ${tomAvgScore} 分")
  40. 2024-03-20 15:33:26,537 INFO storage.BlockManagerInfo: Removed broadcast_8_piece0 on hadoop01:38796 in memory (size: 4.1 KiB, free: 366.2 MiB)
  41. 2024-03-20 15:33:26,547 INFO storage.BlockManagerInfo: Removed broadcast_9_piece0 on hadoop01:38796 in memory (size: 3.4 KiB, free: 366.2 MiB)
  42. 2024-03-20 15:33:26,558 INFO storage.BlockManagerInfo: Removed broadcast_5_piece0 on hadoop01:38796 in memory (size: 3.4 KiB, free: 366.2 MiB)
  43. 2024-03-20 15:33:26,569 INFO storage.BlockManagerInfo: Removed broadcast_4_piece0 on hadoop01:38796 in memory (size: 4.0 KiB, free: 366.2 MiB)
  44. 2024-03-20 15:33:26,578 INFO storage.BlockManagerInfo: Removed broadcast_6_piece0 on hadoop01:38796 in memory (size: 3.5 KiB, free: 366.2 MiB)
  45. 2024-03-20 15:33:26,584 INFO storage.BlockManagerInfo: Removed broadcast_7_piece0 on hadoop01:38796 in memory (size: 3.1 KiB, free: 366.2 MiB)
  46. Tom 同学的总成绩平均分为 30.8 分
  47.  
  48. scala> println("每位同学选修的课程门数如下：")
  49. 每位同学选修的课程门数如下：
  50.  
  51. scala> courseCounts.collect().foreach { case (student, count) =>
  52.      |   println(s"${student} 选修了 ${count} 门课程")
  53.      | }
  54. 2024-03-20 15:33:27,023 INFO spark.SparkContext: Starting job: collect at <console>:24
  55. 2024-03-20 15:33:27,024 INFO scheduler.DAGScheduler: Registering RDD 16 (map at <console>:23) as input to shuffle 3
  56. 2024-03-20 15:33:27,024 INFO scheduler.DAGScheduler: Got job 5 (collect at <console>:24) with 2 output partitions
  57. 2024-03-20 15:33:27,024 INFO scheduler.DAGScheduler: Final stage: ResultStage 9 (collect at <console>:24)
  58. 2024-03-20 15:33:27,024 INFO scheduler.DAGScheduler: Parents of final stage: List(ShuffleMapStage 8)
  59. 2024-03-20 15:33:27,024 INFO scheduler.DAGScheduler: Missing parents: List(ShuffleMapStage 8)
  60. ....
  61. 2024-03-20 15:33:27,092 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 0 ms
  62. 2024-03-20 15:33:27,104 INFO executor.Executor: Finished task 0.0 in stage 9.0 (TID 18). 4478 bytes result sent to driver
  63. 2024-03-20 15:33:27,105 INFO executor.Executor: Finished task 1.0 in stage 9.0 (TID 19). 4470 bytes result sent to driver
  64. 2024-03-20 15:33:27,107 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 9.0 (TID 18) in 21 ms on hadoop01 (executor driver) (1/2)
  65. 2024-03-20 15:33:27,108 INFO scheduler.TaskSetManager: Finished task 1.0 in stage 9.0 (TID 19) in 21 ms on hadoop01 (executor driver) (2/2)
  66. 2024-03-20 15:33:27,108 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 9.0, whose tasks have all completed, from pool
  67. 2024-03-20 15:33:27,108 INFO scheduler.DAGScheduler: ResultStage 9 (collect at <console>:24) finished in 0.028 s
  68. 2024-03-20 15:33:27,108 INFO scheduler.DAGScheduler: Job 5 is finished. Cancelling potential speculative or zombie tasks for this job
  69. 2024-03-20 15:33:27,108 INFO scheduler.TaskSchedulerImpl: Killing all running tasks in stage 9: Stage finished
  70. 2024-03-20 15:33:27,109 INFO scheduler.DAGScheduler: Job 5 finished: collect at <console>:24, took 0.085496 s
  71. Bartholomew 选修了 5 门课程
  72. Lennon 选修了 4 门课程
  73. Joshua 选修了 4 门课程
  74. Tom 选修了 5 门课程
  75. Vic 选修了 3 门课程
  76. Eli 选修了 5 门课程
  77. Alva 选修了 5 门课程
  78. Brady 选修了 5 门课程
  79. Derrick 选修了 6 门课程
  80. Willie 选修了 4 门课程
  81. Bennett 选修了 6 门课程
  82. Boyce 选修了 2 门课程
  83. Elton 选修了 5 门课程
  84. Sidney 选修了 5 门课程
  85. Jay 选修了 6 门课程
  86. Meredith 选修了 4 门课程
  87. Harold 选修了 4 门课程
  88. Jim 选修了 4 门课程
  89. Adonis 选修了 5 门课程
  90. Max 选修了 3 门课程
  91. Abel 选修了 4 门课程
  92. Barton 选修了 1 门课程
  93. Peter 选修了 4 门课程
  94. Matthew 选修了 2 门课程
  95. Alexander 选修了 4 门课程
  96. Donald 选修了 4 门课程
  97. Raymondt 选修了 6 门课程
  98. Devin 选修了 4 门课程
  99. Kerwin 选修了 3 门课程
  100. Borg 选修了 4 门课程
  101. Roy 选修了 6 门课程
  102. Harry 选修了 4 门课程
  103. Abbott 选修了 3 门课程
  104. Miles 选修了 6 门课程
  105. Baron 选修了 6 门课程
  106. Francis 选修了 4 门课程
  107. Lewis 选修了 4 门课程
  108. Aries 选修了 2 门课程
  109. Glenn 选修了 6 门课程
  110. Cleveland 选修了 4 门课程
  111. Mick 选修了 4 门课程
  112. Will 选修了 3 门课程
  113. Henry 选修了 2 门课程
  114. Jesse 选修了 7 门课程
  115. Alvin 选修了 5 门课程
  116. Ivan 选修了 4 门课程
  117. Monroe 选修了 3 门课程
  118. Hobart 选修了 4 门课程
  119. Leo 选修了 5 门课程
  120. Louis 选修了 6 门课程
  121. Randolph 选修了 3 门课程
  122. Sid 选修了 3 门课程
  123. Blair 选修了 4 门课程
  124. Abraham 选修了 3 门课程
  125. Lucien 选修了 5 门课程
  126. Benedict 选修了 6 门课程
  127. Montague 选修了 3 门课程
  128. Giles 选修了 7 门课程
  129. Kerr 选修了 4 门课程
  130. Berg 选修了 4 门课程
  131. Simon 选修了 2 门课程
  132. Lou 选修了 2 门课程
  133. Ronald 选修了 3 门课程
  134. Pete 选修了 3 门课程
  135. Harlan 选修了 6 门课程
  136. Arlen 选修了 4 门课程
  137. Maxwell 选修了 4 门课程
  138. Kennedy 选修了 4 门课程
  139. Bernard 选修了 2 门课程
  140. Spencer 选修了 5 门课程
  141. Andy 选修了 3 门课程
  142. Jeremy 选修了 6 门课程
  143. Alan 选修了 5 门课程
  144. Bruno 选修了 5 门课程
  145. Jerry 选修了 3 门课程
  146. Donahue 选修了 5 门课程
  147. Barry 选修了 5 门课程
  148. Kent 选修了 4 门课程
  149. Frank 选修了 3 门课程
  150. Noah 选修了 4 门课程
  151. Mike 选修了 3 门课程
  152. Tony 选修了 3 门课程
  153. Webb 选修了 7 门课程
  154. Ken 选修了 3 门课程
  155. Philip 选修了 2 门课程
  156. Robin 选修了 4 门课程
  157. Amos 选修了 5 门课程
  158. Chapman 选修了 4 门课程
  159. Valentine 选修了 8 门课程
  160. Angelo 选修了 2 门课程
  161. Boyd 选修了 3 门课程
  162. Chad 选修了 6 门课程
  163. Benjamin 选修了 4 门课程
  164. Allen 选修了 4 门课程
  165. Evan 选修了 3 门课程
  166. Albert 选修了 3 门课程
  167. Alfred 选修了 2 门课程
  168. Newman 选修了 2 门课程
  169. Winston 选修了 4 门课程
  170. Rory 选修了 4 门课程
  171. Dean 选修了 7 门课程
  172. Claude 选修了 2 门课程
  173. Booth 选修了 6 门课程
  174. Channing 选修了 4 门课程
  175. Ward 选修了 4 门课程
  176. Chester 选修了 6 门课程
  177. Webster 选修了 2 门课程
  178. Marshall 选修了 4 门课程
  179. Cliff 选修了 5 门课程
  180. Emmanuel 选修了 3 门课程
  181. Jerome 选修了 3 门课程
  182. Upton 选修了 5 门课程
  183. Corey 选修了 4 门课程
  184. Perry 选修了 5 门课程
  185. Herbert 选修了 3 门课程
  186. Maurice 选修了 2 门课程
  187. Drew 选修了 5 门课程
  188. Brandon 选修了 5 门课程
  189. Adolph 选修了 4 门课程
  190. Levi 选修了 2 门课程
  191. Bing 选修了 6 门课程
  192. Antonio 选修了 3 门课程
  193. Stan 选修了 3 门课程
  194. Les 选修了 6 门课程
  195. Charles 选修了 3 门课程
  196. Clement 选修了 5 门课程
  197. Blithe 选修了 3 门课程
  198. Brian 选修了 6 门课程
  199. Matt 选修了 4 门课程
  200. Archibald 选修了 5 门课程
  201. Horace 选修了 5 门课程
  202. Sebastian 选修了 6 门课程
  203. Verne 选修了 3 门课程
  204. Ford 选修了 3 门课程
  205. Enoch 选修了 3 门课程
  206. Kim 选修了 4 门课程
  207. Conrad 选修了 2 门课程
  208. Marvin 选修了 3 门课程
  209. Michael 选修了 5 门课程
  210. Ernest 选修了 5 门课程
  211. Marsh 选修了 4 门课程
  212. Duke 选修了 4 门课程
  213. Armand 选修了 3 门课程
  214. Lester 选修了 4 门课程
  215. Broderick 选修了 3 门课程
  216. Hayden 选修了 3 门课程
  217. Bertram 选修了 3 门课程
  218. Bart 选修了 5 门课程
  219. Duncann 选修了 5 门课程
  220. Colby 选修了 4 门课程
  221. Bowen 选修了 5 门课程
  222. Elmer 选修了 4 门课程
  223. Elvis 选修了 2 门课程
  224. Adair 选修了 3 门课程
  225. Roderick 选修了 4 门课程
  226. Walter 选修了 4 门课程
  227. Jonathan 选修了 4 门课程
  228. Jo 选修了 5 门课程
  229. Rod 选修了 4 门课程
  230. Scott 选修了 3 门课程
  231. Elliot 选修了 3 门课程
  232. Alvis 选修了 6 门课程
  233. Joseph 选修了 3 门课程
  234. Geoffrey 选修了 4 门课程
  235. Todd 选修了 3 门课程
  236. Wordsworth 选修了 4 门课程
  237. Wright 选修了 4 门课程
  238. Adam 选修了 3 门课程
  239. Sandy 选修了 1 门课程
  240. Ben 选修了 4 门课程
  241. Clyde 选修了 7 门课程
  242. Mark 选修了 7 门课程
  243. Dempsey 选修了 4 门课程
  244. Rock 选修了 6 门课程
  245. Ellis 选修了 4 门课程
  246. Edward 选修了 4 门课程
  247. Eugene 选修了 1 门课程
  248. Samuel 选修了 4 门课程
  249. Gerald 选修了 4 门课程
  250. Luthers 选修了 5 门课程
  251. Virgil 选修了 5 门课程
  252. Bradley 选修了 2 门课程
  253. Dick 选修了 3 门课程
  254. Bevis 选修了 4 门课程
  255. Merlin 选修了 5 门课程
  256. Armstrong 选修了 2 门课程
  257. Ron 选修了 6 门课程
  258. Archer 选修了 5 门课程
  259. Nick 选修了 5 门课程
  260. Hogan 选修了 4 门课程
  261. Len 选修了 5 门课程
  262. Benson 选修了 4 门课程
  263. Colbert 选修了 4 门课程
  264. John 选修了 6 门课程
  265. Saxon 选修了 7 门课程
  266. Marico 选修了 6 门课程
  267. Kevin 选修了 4 门课程
  268. Uriah 选修了 1 门课程
  269. Aldrich 选修了 3 门课程
  270. Jeffrey 选修了 4 门课程
  271. Brook 选修了 4 门课程
  272. Nicholas 选修了 5 门课程
  273. Elijah 选修了 4 门课程
  274. Bill 选修了 2 门课程
  275. Greg 选修了 4 门课程
  276. Payne 选修了 6 门课程
  277. Colin 选修了 5 门课程
  278. Gordon 选修了 4 门课程
  279. Tracy 选修了 3 门课程
  280. Alston 选修了 4 门课程
  281. George 选修了 4 门课程
  282. Griffith 选修了 4 门课程
  283. Andrew 选修了 4 门课程
  284. Egbert 选修了 4 门课程
  285. Bishop 选修了 2 门课程
  286. Beck 选修了 4 门课程
  287. Gilbert 选修了 3 门课程
  288. Phil 选修了 3 门课程
  289. Antony 选修了 5 门课程
  290. Nelson 选修了 5 门课程
  291. Christ 选修了 2 门课程
  292. Bruce 选修了 3 门课程
  293. Rodney 选修了 3 门课程
  294. Boris 选修了 6 门课程
  295. Marlon 选修了 4 门课程
  296. Don 选修了 2 门课程
  297. Aaron 选修了 4 门课程
  298. Sean 选修了 6 门课程
  299. Truman 选修了 3 门课程
  300. Solomon 选修了 5 门课程
  301. Blake 选修了 4 门课程
  302. Christopher 选修了 4 门课程
  303. Clare 选修了 4 门课程
  304. Milo 选修了 2 门课程
  305. Victor 选修了 2 门课程
  306. Nigel 选修了 3 门课程
  307. Jonas 选修了 4 门课程
  308. Jason 选修了 4 门课程
  309. Hilary 选修了 4 门课程
  310. Woodrow 选修了 3 门课程
  311. William 选修了 6 门课程
  312. Dennis 选修了 4 门课程
  313. Jeff 选修了 4 门课程
  314. Dominic 选修了 4 门课程
  315. Merle 选修了 3 门课程
  316. Elroy 选修了 5 门课程
  317. Harvey 选修了 7 门课程
  318. Clark 选修了 6 门课程
  319. Herman 选修了 3 门课程
  320. Bert 选修了 3 门课程
  321. Alger 选修了 5 门课程
  322. Hiram 选修了 6 门课程
  323. Leonard 选修了 2 门课程
  324. Kenneth 选修了 3 门课程
  325. Leopold 选修了 7 门课程
  326. Eric 选修了 4 门课程
  327. Basil 选修了 4 门课程
  328. Martin 选修了 3 门课程
  329. Clarence 选修了 7 门课程
  330. Bernie 选修了 3 门课程
  331. Vincent 选修了 5 门课程
  332. Christian 选修了 2 门课程
  333. Winfred 选修了 3 门课程
  334. Lionel 选修了 4 门课程
  335. Bob 选修了 3 门课程
  336.  
  337. scala> println(s"该系 DataBase 课程共有 ${dbStudents} 人选修")
  338. 该系 DataBase 课程共有 125 人选修
  339. ################################LOG########################################

       学生填写代码以及给出最终结果

（1）      该系总共有多少学生；

val scores = sc.textFile("hdfs://hadoop01:9000/score.txt")


val totalStudents = scores.map(line => line.split(",")(0)).distinct().count()

答案为：265 人

（2）      该系共开设来多少门课程；

val totalCourses = scores.map(line => line.split(",")(1)).distinct().count()

答案为：8门

（3）      Tom同学的总成绩平均分是多少；  

val tomScores = scores.filter(_.startsWith("Tom")).map(_.split(",")(2).toInt)
val tomAvgScore = tomScores.sum() / tomScores.count()

Tom同学的平均分为 30.8分

（4）      求每名同学的选修的课程门数；

val courseCounts = scores.map(line => (line.split(",")(0), 1)).reduceByKey(_ + _)

太多了就不显示了，总共265行

答案共：265行

（5）      该系DataBase课程共有多少人选修；

val dbStudents = scores.filter(_.split(",")(1) == "DataBase").map(_.split(",")(0)).distinct().count()

答案为：125 人

2.spark编程统计客户总消费金额

实验目标：

(1)   掌握数据读取和存储的方法

(2)   掌握RDD的基本操作

实验说明：

现有一份某电商2020年12月份的订单数据文件onlin_retail.csv，记录了每位顾客每笔订单的购物情况，包含三个数据字段，字段说明如下表所示。现需要统计每位客户的总消费金额，并筛选出消费金额在前50名的客户。

实现思路及步骤：

（1）       读取数据并创建RDD

（2）       通过map（）方法分割数据，选择客户编号和订单价格字段组成键值对数据

（3）       使用reduceByKey（）方法计算每位客户的总消费金额

（4）       使用sortBy（）方法对每位客户的总消费金额进行降序排序，取出前50条数

实验过程：

  1. 读取数据并创建 RDD： 从 HDFS 中读取名为 online_retail.txt 的订单数据文件，过滤掉首行（即列名）。

  2. 通过 map() 方法分割数据： 对每一行数据执行 split(",") 操作，将数据切分为字段，并选取顾客编号（第一个字段）和订单价格（第二个字段）作为键值对的键和值。

  3. 使用 reduceByKey() 方法计算总消费金额： 将相同顾客编号的订单价格进行累加，得到每位顾客的总消费金额。

  4. 使用 sortBy() 方法对总消费金额进行降序排序： 将每位顾客的总消费金额进行降序排序，并取出前50名顾客。

  5. 打印结果： 将前50名顾客的顾客编号和总消费金额打印出来。

实现代码（学生填写）：

// (1) 读取数据并创建RDD
val lines = sc.textFile("hdfs://hadoop01:9000/online_retail.txt").filter(!_.startsWith("Customer ID"))

// (2) 通过map()方法分割数据，选择顾客编号和订单价格字段组成键值对数据
val customerSpending = lines.map(line => {
  val fields = line.split(",")
  (fields(0), fields(1).toDouble)
})

// (3) 使用reduceByKey()方法计算每位顾客的总消费金额
val totalSpendingPerCustomer = customerSpending.reduceByKey(_ + _)

// (4) 使用sortBy()方法对每位顾客的总消费金额进行降序排序，取出前50条数据
val top50Customers = totalSpendingPerCustomer.sortBy(_._2, ascending = false).take(50)

// 打印结果
top50Customers.foreach(customer => println(s"顾客ID: ${customer._1} - 总消费金额: ${customer._2}"))

实验记录：

  1. scala> val lines = sc.textFile("hdfs://hadoop01:9000/online_retail.txt").filter(!_.startsWith("Customer ID"))
  2. 2024-03-25 15:32:11,083 INFO memory.MemoryStore: Block broadcast_2 stored as values in memory (estimated size 389.1 KiB, free 365.5 MiB)
  3. 2024-03-25 15:32:11,122 INFO memory.MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 40.2 KiB, free 365.5 MiB)
  4. 2024-03-25 15:32:11,126 INFO storage.BlockManagerInfo: Added broadcast_2_piece0 in memory on hadoop01:40652 (size: 40.2 KiB, free: 366.2 MiB)
  5. 2024-03-25 15:32:11,128 INFO spark.SparkContext: Created broadcast 2 from textFile at <console>:23
  6. lines: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[9] at filter at <console>:23
  7.  
  8. scala> val customerSpending = lines.map(line => {
  9.      |   val fields = line.split(",")
  10.      |   (fields(0), fields(1).toDouble)
  11.      | })
  12. customerSpending: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[10] at map at <console>:23
  13.  
  14. scala> val totalSpendingPerCustomer = customerSpending.reduceByKey(_ + _)
  15. 2024-03-25 15:32:22,505 INFO mapred.FileInputFormat: Total input files to process : 1
  16. totalSpendingPerCustomer: org.apache.spark.rdd.RDD[(String, Double)] = ShuffledRDD[11] at reduceByKey at <console>:23
  17.  
  18. scala> val top50Customers = totalSpendingPerCustomer.sortBy(_._2, ascending = false).take(50)
  19. 2024-03-25 15:32:26,850 INFO spark.SparkContext: Starting job: sortBy at <console>:23
  20. 2024-03-25 15:32:26,856 INFO scheduler.DAGScheduler: Registering RDD 10 (map at <console>:23) as input to shuffle 1
  21. 2024-03-25 15:32:26,857 INFO scheduler.DAGScheduler: Got job 1 (sortBy at <console>:23) with 2 output partitions
  22. ......
  23. 2024-03-25 15:32:27,852 INFO executor.Executor: Finished task 0.0 in stage 6.0 (TID 8). 3276 bytes result sent to driver
  24. 2024-03-25 15:32:27,854 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 6.0 (TID 8) in 69 ms on hadoop01 (executor driver) (1/1)
  25. 2024-03-25 15:32:27,854 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 6.0, whose tasks have all completed, from pool
  26. 2024-03-25 15:32:27,855 INFO scheduler.DAGScheduler: ResultStage 6 (take at <console>:23) finished in 0.090 s
  27. 2024-03-25 15:32:27,856 INFO scheduler.DAGScheduler: Job 2 is finished. Cancelling potential speculative or zombie tasks for this job
  28. 2024-03-25 15:32:27,856 INFO scheduler.TaskSchedulerImpl: Killing all running tasks in stage 6: Stage finished
  29. 2024-03-25 15:32:27,856 INFO scheduler.DAGScheduler: Job 2 finished: take at <console>:23, took 0.211368 s
  30. top50Customers: Array[(String, Double)] = Array(("",174463.66000000347), (12748,1618.1500000000015), (14911,1573.1600000000014), (17850,1176.2299999999982), (17841,1073.1299999999997), (14606,828.5199999999996), (16607,726.3199999999999), (14527,666.1399999999996), (17340,613.5099999999999), (15311,582.7899999999997), (15044,545.4599999999998), (13174,519.7500000000001), (14667,506.5499999999998), (15727,456.44999999999993), (17961,414.82000000000005), (14030,413.75), (18116,410.97999999999973), (15039,404.6699999999998), (16873,401.4299999999999), (18118,389.7099999999998), (15574,386.2499999999999), (14180,382.24999999999994), (16713,377.51999999999987), (18055,377.3799999999999), (14505,373.29999999999956), (15498,369.66999999999996), (15808,366.299999999999...
  31.  
  32. scala> top50Customers.foreach(customer => println(s"顾客ID: ${customer._1} - 总消费金额: ${customer._2}"))
  33. 2024-03-25 15:32:32,608 INFO storage.BlockManagerInfo: Removed broadcast_4_piece0 on hadoop01:40652 in memory (size: 4.0 KiB, free: 366.2 MiB)
  34. 顾客ID:  - 总消费金额: 174463.66000000347
  35. 顾客ID: 12748 - 总消费金额: 1618.1500000000015
  36. 顾客ID: 14911 - 总消费金额: 1573.1600000000014
  37. 顾客ID: 17850 - 总消费金额: 1176.2299999999982
  38. 顾客ID: 17841 - 总消费金额: 1073.1299999999997
  39. 顾客ID: 14606 - 总消费金额: 828.5199999999996
  40. 顾客ID: 16607 - 总消费金额: 726.3199999999999
  41. 顾客ID: 14527 - 总消费金额: 666.1399999999996
  42. 顾客ID: 17340 - 总消费金额: 613.5099999999999
  43. 顾客ID: 15311 - 总消费金额: 582.7899999999997
  44. 顾客ID: 15044 - 总消费金额: 545.4599999999998
  45. 顾客ID: 13174 - 总消费金额: 519.7500000000001
  46. 顾客ID: 14667 - 总消费金额: 506.5499999999998
  47. 顾客ID: 15727 - 总消费金额: 456.44999999999993
  48. 顾客ID: 17961 - 总消费金额: 414.82000000000005
  49. 顾客ID: 14030 - 总消费金额: 413.75
  50. 顾客ID: 18116 - 总消费金额: 410.97999999999973
  51. 顾客ID: 15039 - 总消费金额: 404.6699999999998
  52. 顾客ID: 16873 - 总消费金额: 401.4299999999999
  53. 顾客ID: 18118 - 总消费金额: 389.7099999999998
  54. 顾客ID: 15574 - 总消费金额: 386.2499999999999
  55. 顾客ID: 14180 - 总消费金额: 382.24999999999994
  56. 顾客ID: 16713 - 总消费金额: 377.51999999999987
  57. 顾客ID: 18055 - 总消费金额: 377.3799999999999
  58. 顾客ID: 14505 - 总消费金额: 373.29999999999956
  59. 顾客ID: 15498 - 总消费金额: 369.66999999999996
  60. 顾客ID: 15808 - 总消费金额: 366.2999999999998
  61. 顾客ID: 15570 - 总消费金额: 363.50999999999993
  62. 顾客ID: 12567 - 总消费金额: 360.3999999999999
  63. 顾客ID: 17341 - 总消费金额: 343.92999999999995
  64. 顾客ID: 16003 - 总消费金额: 339.14
  65. 顾客ID: 15159 - 总消费金额: 338.5599999999999
  66. 顾客ID: 12471 - 总消费金额: 337.2599999999997
  67. 顾客ID: 15640 - 总消费金额: 335.16999999999996
  68. 顾客ID: 12647 - 总消费金额: 328.49999999999994
  69. 顾客ID: 15514 - 总消费金额: 326.68
  70. 顾客ID: 17377 - 总消费金额: 320.38999999999993
  71. 顾客ID: 16782 - 总消费金额: 310.3399999999999
  72. 顾客ID: 15998 - 总消费金额: 307.53999999999985
  73. 顾客ID: 17827 - 总消费金额: 305.8199999999997
  74. 顾客ID: 14415 - 总消费金额: 305.21999999999974
  75. 顾客ID: 14573 - 总消费金额: 304.6999999999999
  76. 顾客ID: 13564 - 总消费金额: 300.01
  77. 顾客ID: 17591 - 总消费金额: 297.5499999999999
  78. 顾客ID: 13145 - 总消费金额: 295.0
  79. 顾客ID: 15061 - 总消费金额: 293.1500000000001
  80. 顾客ID: 14083 - 总消费金额: 287.68999999999994
  81. 顾客ID: 16274 - 总消费金额: 286.5999999999998
  82. 顾客ID: 14723 - 总消费金额: 279.93
  83. 顾客ID: 14733 - 总消费金额: 279.06

实验结果：

3.spark编程统计各城市的平均气温

实验目标：

(1)   掌握RDD创建方法

(2)   掌握map，groupby，mapvalues，reduce方法的使用

实验说明：

现有一份各城市的温度数据文件avgTemperature.txt,数据如下表所示，记录了某段时间范围内各城市每天的温度，文件中每一行数据分别表示城市名和温度，现要求用spark编程计算出各城市的平均气温。

实现思路及步骤：

（1）       通过textFile（）方法读取数据创建RDD

（2）       使用map（）方法将数据输入数据按制表符进行分割，并转化成（城市，温度）的形式

（3）       使用groupBy（）方法按城市分组，得到每个城市对应的所欲温度。

（4）       使用mapValues（）和reduce（）方法计算各城市的平均气温

实验过程：

  1. 通过textFile()方法读取数据创建RDD： 使用 SparkContext 的 textFile() 方法从 HDFS 中的 avgTemperature.txt 文件读取数据，并创建一个包含文件中每一行的 RDD。

  2. 使用map()方法将数据按制表符进行分割，并转化成(城市，温度)的形式： 对 RDD 中的每一行数据执行 map() 操作，将每行数据按制表符进行分割，并将城市名和温度值组成键值对。

  3. 使用groupBy()方法按城市分组，得到每个城市对应的所有温度： 使用 groupBy() 方法将键值对按城市分组，得到每个城市对应的所有温度的 Iterable。

  4. 使用mapValues()和reduce()方法计算各城市的平均气温： 对每个城市的温度集合使用 mapValues() 方法计算平均气温，然后使用 reduce() 方法对温度进行求和，并除以温度的数量，得到平均值。

  5. 打印结果： 使用 foreach() 方法遍历每个城市的平均气温，并将结果打印出来，格式为 "城市：平均气温"。

实现代码（学生填写）：

// (1) 通过textFile()方法读取数据创建RDD
val lines = sc.textFile("hdfs://hadoop01:9000/avgTemperature.txt")

// (2) 使用map()方法将数据按制表符进行分割，并转化成(城市，温度)的形式
val cityTemperatures = lines.map(line => {
  val fields = line.split("\t")
  (fields(0), fields(1).toDouble)
})

// (3) 使用groupBy()方法按城市分组，得到每个城市对应的所有温度
val cityTemperatureGroups = cityTemperatures.groupByKey()

// (4) 使用mapValues()和reduce()方法计算各城市的平均气温
val averageTemperatures = cityTemperatureGroups.mapValues(temperatures => temperatures.reduce(_ + _) / temperatures.size)

// 打印结果
averageTemperatures.foreach(city => println(s"城市：${city._1} 平均气温：${city._2}"))

实验记录：

  1. scala> val lines = sc.textFile("hdfs://hadoop01:9000/avgTemperature.txt")
  2. 2024-03-25 15:40:46,632 INFO memory.MemoryStore: Block broadcast_7 stored as values in memory (estimated size 389.1 KiB, free 365.1 MiB)
  3. 2024-03-25 15:40:46,671 INFO memory.MemoryStore: Block broadcast_7_piece0 stored as bytes in memory (estimated size 40.2 KiB, free 365.0 MiB)
  4. 2024-03-25 15:40:46,673 INFO storage.BlockManagerInfo: Added broadcast_7_piece0 in memory on hadoop01:40652 (size: 40.2 KiB, free: 366.2 MiB)
  5. 2024-03-25 15:40:46,675 INFO spark.SparkContext: Created broadcast 7 from textFile at <console>:23
  6. lines: org.apache.spark.rdd.RDD[String] = hdfs://hadoop01:9000/avgTemperature.txt MapPartitionsRDD[18] at textFile at <console>:23
  7.  
  8. scala> val cityTemperatures = lines.map(line => {
  9.      |   val fields = line.split("\t")
  10.      |   (fields(0), fields(1).toDouble)
  11.      | })
  12. cityTemperatures: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[19] at map at <console>:23
  13.  
  14. scala> val cityTemperatureGroups = cityTemperatures.groupByKey()
  15. 2024-03-25 15:40:55,462 INFO mapred.FileInputFormat: Total input files to process : 1
  16. cityTemperatureGroups: org.apache.spark.rdd.RDD[(String, Iterable[Double])] = ShuffledRDD[20] at groupByKey at <console>:23
  17.  
  18. scala> val averageTemperatures = cityTemperatureGroups.mapValues(temperatures => temperatures.reduce(_ + _) / temperatures.size)
  19. averageTemperatures: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[21] at mapValues at <console>:23
  20.  
  21. scala> averageTemperatures.foreach(city => println(s"城市：${city._1} 平均气温：${city._2}"))
  22. 2024-03-25 15:41:05,012 INFO spark.SparkContext: Starting job: foreach at <console>:24
  23. 2024-03-25 15:41:05,017 INFO scheduler.DAGScheduler: Registering RDD 19 (map at <console>:23) as input to shuffle 3
  24. 2024-03-25 15:41:05,018 INFO scheduler.DAGScheduler: Got job 3 (foreach at <console>:24) with 2 output partitions
  25. 2024-03-25 15:41:05,018 INFO scheduler.DAGScheduler: Final stage: ResultStage 8 (foreach at <console>:24)
  26. 2024-03-25 15:41:05,019 INFO scheduler.DAGScheduler: Parents of final stage: List(ShuffleMapStage 7)
  27. ....
  28. 2024-03-25 15:41:05,234 INFO storage.ShuffleBlockFetcherIterator: Getting 2 (177.0 B) non-empty blocks including 2 (177.0 B) local and 0 (0.0 B) host-local and 0 (0.0 B) push-merged-local and 0 (0.0 B) remote blocks
  29. 2024-03-25 15:41:05,234 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 3 ms
  30. 2024-03-25 15:41:05,234 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 3 ms
  31. 城市：shenzhen 平均气温：32.6
  32. 城市：guangzhou 平均气温：32.43333333333333
  33. 城市：shanghai 平均气温：29.3
  34. 城市：beijing 平均气温：27.86666666666667
  35.

实验结果：

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/663256
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号