2023_Spark_实行十一：RDD基础算子操作 - ToB企服应用市场:ToB评测及商务社交产业平台

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

复制代码

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

复制代码

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

复制代码

// 从array中创建RDD
val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)
distData.foreach(println)
// 读取文件创建RDD
val lines = sc.textFile("D:\\PycharmProjects\\2024\\pyspark\\datas\\word.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)
println(totalLength)
// 数据持久化
lineLengths.persist()
print(lineLengths.reduce((a, b) => a + b))
// 对象的函数
object MyFunctions {
def func1(s: String): String = { s"打印RDD中的字符串，包含的字符串有: $s" }
}
val myRdd = lines.flatMap(lines => lines.split(" "))
myRdd.map(MyFunctions.func1).foreach(println)
import org.apache.spark.rdd.RDD
// 类的函数
class MyClass extends Serializable {
def func1(s: String): String = { f"在MyClass类中，打印RDD中的字符串，包含的字符串有: $s" }
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}
val f1 = new MyClass()
f1.doStuff(myRdd).foreach(println)
// 类的应用
class MyClass2 extends Serializable {
val field = "你好，测试案例..."
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}
val f2 = new MyClass2()
f2.doStuff(myRdd).foreach(println)
// Pair RDD应用
val lines = sc.textFile("D:\\PycharmProjects\\2024\\pyspark\\datas\\word.txt")
val pairs = lines.flatMap(_.split(" ")).map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)
// 交换键和值的位置
val swappedCounts = counts.map(_.swap)
// 先根据值排序（降序），然后根据键排序（升序）
val sortedByValueThenKeyDesc = swappedCounts.sortByKey(ascending = false)
val CountsDescondvalue = sortedByValueThenKeyDesc .map(_.swap)
CountsDescondvalue .collect()
// 广播变量 Broadcast Variables
val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar.value
val accum = sc.longAccumulator("My Accumulator")
sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
accum.value

复制代码

package test
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
/**
* @projectName GNUSpark20204
* @package test
* @className test.RDD_spark341
* @description ${description}
* @author pblh123
* @date 2024/9/26 23:08
* @version 1.0
*
*/
object RDD_spark341 extends App {
// 创建SparkSession sparkcontext
val spark = SparkSession.builder
.appName("RDD_spark341")
.master("local[2]")
.getOrCreate()
val sc: SparkContext = spark.sparkContext
// spark代码主体
// 从array中创建RDD
val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)
distData.foreach(println)
// 读取文件创建RDD
val lines = sc.textFile("D:\\PycharmProjects\\2024\\pyspark\\datas\\word.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)
println(totalLength)
// 数据持久化
lineLengths.persist()
print(lineLengths.reduce((a, b) => a + b))
// 对象的函数
object MyFunctions {
def func1(s: String): String = {
s"打印RDD中的字符串，包含的字符串有: $s"
}
}
val myRdd = lines.flatMap(lines => lines.split(" "))
myRdd.map(MyFunctions.func1).foreach(println)
import org.apache.spark.rdd.RDD
// 类的函数
class MyClass extends Serializable {
def func1(s: String): String = {
f"在MyClass类中，打印RDD中的字符串，包含的字符串有: $s"
}
def doStuff(rdd: RDD[String]): RDD[String] = {
rdd.map(func1)
}
}
val f1 = new MyClass()
f1.doStuff(myRdd).foreach(println)
// 类的应用
class MyClass2 extends Serializable {
val field = "你好，测试案例..."
def doStuff(rdd: RDD[String]): RDD[String] = {
rdd.map(x => field + x)
}
}
val f2 = new MyClass2()
f2.doStuff(myRdd).foreach(println)
// Pair RDD应用
val pairs = lines.flatMap(_.split(" ")).map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)
// 交换键和值的位置
val swappedCounts = counts.map(_.swap)
// 先根据值排序（降序），然后根据键排序（升序）
val sortedByValueThenKeyDesc = swappedCounts.sortByKey(ascending = false)
val CountsDescondvalue = sortedByValueThenKeyDesc.map(_.swap)
println(CountsDescondvalue.collect())
// 广播变量 Broadcast Variables
val broadcastVar = sc.broadcast(Array(1, 2, 3))
println(broadcastVar.value)
val accum = sc.longAccumulator("My Accumulator")
sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
println(accum.value)
// 关闭sparkSesssion sparkcontext
sc.stop()
spark.stop()
}

复制代码