2023_Spark_实行十一：RDD基础算子操作

一、RDD的练习可以使用两种方式

• 使用Shell
  
• 使用IDEA
  

二、使用Shell练习RDD

当你打开 Spark 的交互式命令行界面（也就是 Spark shell）的时间，它已经自动为你准备好了一个叫做 sc 的特别对象，这个对象是用来和 Spark 集群沟通的。你不需要，也不应该本身再创建一个如许的对象。
如果你想告诉 Spark 用哪个计算机大概计算机集群来执行你的命令，可以通过 --master 这个选项来设置。比如，你想在当地计算机上只用四个核心来运行，就可以在命令里加上 --master local[4]。

1. $ ./bin/spark-shell --master local[4]

复制代码

如果你有一些本身的代码打包成了 JAR 文件，想要在 Spark shell 里用，可以通过 --jars 选项，反面跟上你的 JAR 文件名，用逗号分隔，来把它们加入到可以识别的路径里。

1. $ ./bin/spark-shell --master local[4] --jars code.jar

复制代码

此外，如果你需要一些额外的库大概 Spark 的扩展包，可以通过 --packages 选项，反面跟上这些库的 Maven 坐标（一种常用的依靠管理方式），用逗号分隔，来添加它们。假设你需要的包是 org.apache.spark:spark-mllib_2.13:3.4.1，这是Spark的呆板学习库。

1. $ ./bin/spark-shell --master local[4] --packages "org.apache.spark:spark-mllib_2.13:3.4.1"

复制代码

简单来说，这些选项就是让你告诉 Spark 怎么运行你的代码，以及在哪里找到运行代码所需要的资源。

RDD基础

1. // 从array中创建RDD
2. val data = Array(1, 2, 3, 4, 5)
3. val distData = sc.parallelize(data)
4. distData.foreach(println)
7. // 读取文件创建RDD
8. val lines = sc.textFile("D:\\PycharmProjects\\2024\\pyspark\\datas\\word.txt")
9. val lineLengths = lines.map(s => s.length)
10. val totalLength = lineLengths.reduce((a, b) => a + b)
11. println(totalLength)
13. // 数据持久化
14. lineLengths.persist()
15. print(lineLengths.reduce((a, b) => a + b))
18. // 对象的函数
19. object MyFunctions {
20.   def func1(s: String): String = { s"打印RDD中的字符串，包含的字符串有: $s" }
21. }
22. val myRdd = lines.flatMap(lines => lines.split(" "))
23. myRdd.map(MyFunctions.func1).foreach(println)
26. import org.apache.spark.rdd.RDD
27. // 类的函数
28. class MyClass extends Serializable {
29.   def func1(s: String): String = { f"在MyClass类中，打印RDD中的字符串，包含的字符串有: $s" }
30.   def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
31. }
33. val f1 = new MyClass()
34. f1.doStuff(myRdd).foreach(println)
37. // 类的应用
38. class MyClass2 extends Serializable {
39.   val field = "你好，测试案例..."
40.   def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
41. }
42. val f2 = new MyClass2()
43. f2.doStuff(myRdd).foreach(println)
45. // Pair RDD应用
46. val lines = sc.textFile("D:\\PycharmProjects\\2024\\pyspark\\datas\\word.txt")
47. val pairs = lines.flatMap(_.split(" ")).map(s => (s, 1))
48. val counts = pairs.reduceByKey((a, b) => a + b)
49. // 交换键和值的位置
50. val swappedCounts = counts.map(_.swap)
51. // 先根据值排序（降序），然后根据键排序（升序）
52. val sortedByValueThenKeyDesc = swappedCounts.sortByKey(ascending = false)
53. val CountsDescondvalue = sortedByValueThenKeyDesc .map(_.swap)
54. CountsDescondvalue .collect()
57. // 广播变量 Broadcast Variables
58. val broadcastVar = sc.broadcast(Array(1, 2, 3))
59. broadcastVar.value
61. val accum = sc.longAccumulator("My Accumulator")
62. sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
63. accum.value

复制代码

三、使用IDEA练习RDD

基于Spark3.4.1，IDEA练习基础的RDD

1. package test
3. import org.apache.spark.SparkContext
4. import org.apache.spark.sql.SparkSession
6. /**
7. * @projectName GNUSpark20204  
8. * @package test  
9. * @className test.RDD_spark341  
10. * @description ${description}  
11. * @author pblh123
12. * @date 2024/9/26 23:08
13. * @version 1.0
14. *
15. */
16.    
17. object RDD_spark341 extends App {
19. //  创建SparkSession sparkcontext
20.   val spark = SparkSession.builder
21.     .appName("RDD_spark341")
22.     .master("local[2]")
23.     .getOrCreate()
24.   val sc: SparkContext = spark.sparkContext
26. //  spark代码主体
27.   // 从array中创建RDD
28.     val data = Array(1, 2, 3, 4, 5)
29.     val distData = sc.parallelize(data)
30.     distData.foreach(println)
33.     // 读取文件创建RDD
34.     val lines = sc.textFile("D:\\PycharmProjects\\2024\\pyspark\\datas\\word.txt")
35.     val lineLengths = lines.map(s => s.length)
36.     val totalLength = lineLengths.reduce((a, b) => a + b)
37.     println(totalLength)
39.     // 数据持久化
40.     lineLengths.persist()
41.     print(lineLengths.reduce((a, b) => a + b))
44.     // 对象的函数
45.     object MyFunctions {
46.       def func1(s: String): String = {
47.         s"打印RDD中的字符串，包含的字符串有: $s"
48.       }
49.     }
51.     val myRdd = lines.flatMap(lines => lines.split(" "))
52.     myRdd.map(MyFunctions.func1).foreach(println)
55.     import org.apache.spark.rdd.RDD
57.     // 类的函数
58.     class MyClass extends Serializable {
59.       def func1(s: String): String = {
60.         f"在MyClass类中，打印RDD中的字符串，包含的字符串有: $s"
61.       }
63.       def doStuff(rdd: RDD[String]): RDD[String] = {
64.         rdd.map(func1)
65.       }
66.     }
68.     val f1 = new MyClass()
69.     f1.doStuff(myRdd).foreach(println)
72.     // 类的应用
73.     class MyClass2 extends Serializable {
74.       val field = "你好，测试案例..."
76.       def doStuff(rdd: RDD[String]): RDD[String] = {
77.         rdd.map(x => field + x)
78.       }
79.     }
81.     val f2 = new MyClass2()
82.     f2.doStuff(myRdd).foreach(println)
84.     // Pair RDD应用
85.     val pairs = lines.flatMap(_.split(" ")).map(s => (s, 1))
86.     val counts = pairs.reduceByKey((a, b) => a + b)
87.     // 交换键和值的位置
88.     val swappedCounts = counts.map(_.swap)
89.     // 先根据值排序（降序），然后根据键排序（升序）
90.     val sortedByValueThenKeyDesc = swappedCounts.sortByKey(ascending = false)
91.     val CountsDescondvalue = sortedByValueThenKeyDesc.map(_.swap)
92.     println(CountsDescondvalue.collect())
95.     // 广播变量 Broadcast Variables
96.     val broadcastVar = sc.broadcast(Array(1, 2, 3))
97.     println(broadcastVar.value)
99.     val accum = sc.longAccumulator("My Accumulator")
100.     sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
101.     println(accum.value)
103. //  关闭sparkSesssion sparkcontext
104.   sc.stop()
105.   spark.stop()
107. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。