23) sortByKey
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名称和运行模式
val conf = new SparkConf().setAppName("SortByKeyExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建一个 (K, V) 格式的 RDD
val dataRDD1 = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
// 按键升序排序
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(true)
// 按键降序排序
val sortRDD2: RDD[(String, Int)] = dataRDD1.sortByKey(false)
// 打印升序排序结果
println("Ascending order:")
sortRDD1.collect().foreach(println)
// 打印降序排序结果
println("Descending order:")
sortRDD2.collect().foreach(println)
// 停止 SparkContext，释放资源
sc.stop()
}
}
- Join
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名称和运行模式
val conf = new SparkConf().setAppName("JoinExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建第一个 (K, V) 格式的 RDD
val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
// 创建第二个 (K, V) 格式的 RDD
val rdd1: RDD[(Int, Int)] = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))
// 对两个 RDD 进行 join 操作
val joinedRDD: RDD[(Int, (String, Int))] = rdd.join(rdd1)
// 收集结果并打印
joinedRDD.collect().foreach(println)
// 停止 SparkContext，释放资源
sc.stop()
}
}
- leftOuterJoin
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象
val conf = new SparkConf().setAppName("LeftOuterJoinExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建第一个 (K, V) 形式的 RDD
val dataRDD1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 4)))
// 创建第二个 (K, V) 形式的 RDD
val dataRDD2: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
// 对两个 RDD 进行左外连接操作
val rdd: RDD[(String, (Int, Option[Int]))] = dataRDD1.leftOuterJoin(dataRDD2)
// 打印左外连接结果
println("左外连接结果:")
rdd.collect().foreach(println)
// 停止 SparkContext
sc.stop()
}
}
- Cogroup
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object CogroupExample {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象
val conf = new SparkConf().setAppName("CogroupExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建第一个 (K, V) 形式的 RDD
val dataRDD1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("c", 3)))
// 创建第二个 (K, V) 形式的 RDD
val dataRDD2: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("c", 2), ("c", 3)))
// 对两个 RDD 进行 cogroup 操作
val value: RDD[(String, (Iterable[Int], Iterable[Int]))] = dataRDD1.cogroup(dataRDD2)
// 打印 cogroup 操作结果
println("cogroup 操作结果:")
value.collect().foreach(println)
// 停止 SparkContext
sc.stop()
}
}
Spark-Core编程（五）
1) reduce
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名称和本地运行模式
val conf = new SparkConf().setAppName("RDDReduceExample").setMaster("local[*]")
// 根据配置创建 SparkContext 对象
val sc = new SparkContext(conf)
// 使用 makeRDD 方法将列表转换为 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 使用 reduce 方法对 RDD 元素进行聚合，这里是求和操作
val reduceResult: Int = rdd.reduce(_ + _)
// 打印聚合结果
println(reduceResult)
// 停止 SparkContext 以释放资源
sc.stop()
}
}
2) collect
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名称和运行模式为本地模式
val conf = new SparkConf().setAppName("RDDCollectExample").setMaster("local[*]")
// 创建 SparkContext 对象，用于与 Spark 集群交互
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 使用 collect 方法将 RDD 中的所有元素以数组形式返回到驱动程序
val collectedArray: Array[Int] = rdd.collect()
// 打印数组中的每个元素
collectedArray.foreach(println)
// 停止 SparkContext，释放资源
sc.stop()
}
}
3) foreach
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名称和运行模式为本地模式
val conf = new SparkConf().setAppName("RDDForeachExample").setMaster("local[*]")
// 创建 SparkContext 对象，用于与 Spark 集群交互
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 分布式遍历 RDD 中的每一个元素，调用 println 函数
rdd.foreach(println)
// 停止 SparkContext，释放资源
sc.stop()
}
}
4) count
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww{
def main(args: Array[String]): Unit = {
// 创建 Spark 配置对象，设置应用名称和运行模式
val conf = new SparkConf().setAppName("RDDCountExample").setMaster("local[*]")
// 创建 Spark 上下文对象
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 使用 count 方法统计 RDD 中元素的个数
val countResult: Long = rdd.count()
// 打印统计结果
println(countResult)
// 停止 Spark 上下文
sc.stop()
}
}
5) first
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名和运行模式
val conf = new SparkConf().setAppName("FirstExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 使用 first 方法获取 RDD 的第一个元素
val firstResult: Int = rdd.first()
// 打印获取到的第一个元素
println(firstResult)
// 停止 SparkContext
sc.stop()
}
}
6) take
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww{
def main(args: Array[String]): Unit = {
// 创建 Spark 配置对象，设置应用名称和运行模式
val conf = new SparkConf().setAppName("RDDTakeExample").setMaster("local[*]")
// 创建 Spark 上下文对象
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 使用 take 方法获取 RDD 的前 2 个元素
val takeResult: Array[Int] = rdd.take(2)
// 遍历并打印获取到的元素
takeResult.foreach(println)
// 停止 Spark 上下文
sc.stop()
}
}
7) takeOrdered
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 Spark 配置对象，设置应用名称和运行模式
val conf = new SparkConf().setAppName("RDDTakeOrderedExample").setMaster("local[*]")
// 创建 Spark 上下文对象
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 3, 2, 4))
// 使用 takeOrdered 方法获取 RDD 排序后的前 2 个元素
val result: Array[Int] = rdd.takeOrdered(2)
// 遍历并打印获取到的元素
result.foreach(println)
// 停止 Spark 上下文
sc.stop()
}
}
- Aggregate
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名和运行模式
val conf = new SparkConf().setAppName("RDDAggregateExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD，设置分区数为 8
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 8)
// 将该 RDD 所有元素相加得到结果，初始值为 0
val result1: Int = rdd.aggregate(0)(_ + _, _ + _)
// 将该 RDD 所有元素相加得到结果，初始值为 10
val result2: Int = rdd.aggregate(10)(_ + _, _ + _)
// 打印结果
println(result1)
println("**********")
println(result2)
// 停止 SparkContext
sc.stop()
}
}
9) fold
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名和运行模式
val conf = new SparkConf().setAppName("RDDFoldExample").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 使用 fold 方法进行折叠操作，初始值为 0
val foldResult: Int = rdd.fold(0)(_ + _)
// 打印折叠操作的结果
println(foldResult)
// 停止 SparkContext
sc.stop()
}
}
10) countByKey
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 Spark 配置对象，设置应用名称和运行模式
val conf = new SparkConf().setAppName("RDDCountByKeyExample").setMaster("local[*]")
// 创建 Spark 上下文对象
val sc = new SparkContext(conf)
// 创建一个包含键值对的 RDD
val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))
// 使用 countByKey 方法统计每种 key 的个数
val result: collection.Map[Int, Long] = rdd.countByKey()
// 打印统计结果
println(result)
// 停止 Spark 上下文
sc.stop()
}
}
- save 相关算子
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Cww {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用名称和运行模式为本地模式
val conf = new SparkConf().setAppName("RDDSaveOperatorsExample").setMaster("local[*]")
// 创建 SparkContext 对象，用于与 Spark 集群交互
val sc = new SparkContext(conf)
// 创建一个包含整数的 RDD
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
// 保存成 Text 文件
rdd.saveAsTextFile("Spark-core/output/output")
// 序列化成对象保存到文件
rdd.saveAsObjectFile("Spark-core/output/output1")
// 停止 SparkContext，释放资源
sc.stop()
}
}
累加器
实现原理
累加器用来把 Executor 端变量信息聚合到 Driver 端。在 Driver 程序中定义的变量，在
Executor 端的每个 Task 都会得到这个变量的一份新的副本，每个 task 更新这些副本的值后，传回 Driver 端进行 merge。
val rdd = sparkContext.makeRDD(List(1,2,3,4,5))
// 声明累加器
var sum = sparkContext.longAccumulator("sum");
rdd.foreach(
num => {
// 使用累加器
sum.add(num)
}
)
// 获取累加器的值
println("sum = " + sum.value)
自定义累加器实现wordcount：
创建自定义累加器：
class WordCountAccumulator extends AccumulatorV2[String,mutable.Map[String,Long]] {
var map:mutable.Map[String,Long] = mutable.Map()
override def isZero: Boolean = map.isEmpty
override def copy(): AccumulatorV2[String, mutable.Map[String,Long]] = new WordCountAccumulator
override def reset(): Unit = map.clear()
override def add(v: String): Unit = {
map(v) = map.getOrElse(v,0L)+1L
}
override def merge(other: AccumulatorV2[String, mutable.Map[String,Long]
]): Unit = {
val map1 = map
val map2 = other.value
map = map1.foldLeft(map2)(
(innerMap,kv)=>{
innerMap(kv._1) = innerMap.getOrElse(kv._1,0L)+kv._2
innerMap
}
)
}
override def value: mutable.Map[String,Long] = map
}
调用自定义累加器：
val rdd = sparkContext.makeRDD(
List("spark","scala","spark hadoop","hadoop")
)
val acc = new WordCountAccumulator
sparkContext.register(acc)
rdd.flatMap(_.split(" ")).foreach(
word=>acc.add(word)
)
println(acc.value)
广播变量
实现原理
广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个
或多个 Spark 操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，
广播变量用起来都很顺手。在多个并行操作中使用同一个变量，但是 Spark 会为每个任务
分别发送。
val rdd1 = sparkContext.makeRDD(List( ("a",1), ("b", 2), ("c", 3), ("d", 4) ),4)
val list = List( ("a",4), ("b", 5), ("c", 6), ("d", 7))
val broadcast :Broadcast[List[(String,Int)]] = sparkContext.broadcast(list)
val resultRDD :RDD[(String,(Int,Int))] = rdd1.map{
case (key,num)=> {
var num2 = 0
for((k,v)<-broadcast.value){
if(k == key) {
num2 = v
}
}
(key,(num,num2))
}
}
resultRDD.collect().foreach(println)
sparkContext.stop()
}
