大数据-84 Spark RDD创建全攻略：从聚集、文件到转换利用详解

点一下关注吧！！！非常感谢！！连续更新！！！

🚀 AI篇连续更新中！（恒久更新）

AI炼丹日志-31- 千呼万唤始出来 GPT-5 发布！“快的模子 + 深度思索模子 + 实时路由”，连续打造实用AI工具指南！📐🤖
💻 Java篇正式开启！（300篇）

现在2025年09月01日更新到：
Java-113 深入浅出 MySQL 扩容全攻略：触发条件、迁徙方案与性能优化
MyBatis 已完结，Spring 已完结，Nginx已完结，Tomcat已完结，分布式服务正在更新！深入浅出助你打牢底子！
📊 大数据板块已完成多项干货更新（300篇）：

包罗 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项焦点组件，覆盖离线+实时数仓全栈！
大数据-278 Spark MLib - 底子先容 呆板学习算法 梯度提拔树 GBDT案例 详解

章节内容

上节我们完成了如下的内容：

• RDD的先容
  
• RDD的特点、特点先容
  
• Spark 编程模子的先容
  

RDD 的创建

SparkContext

SparkContext是Spark应用步调的焦点组件，也是编写Spark步调时须要用到的第一个类。作为Spark的告急入口点，它负担着与整个集群交互的告急职责。
焦点功能与定位

• 客户端与服务端模子
  
  
  
  • 假如把Spark集群比作服务端，那么Driver步调就是客户端，而SparkContext就是这个客户端的焦点引擎
    
  • 比方：当提交一个Spark作业时，Driver步调中的SparkContext会负责与集群管理器（如YARN、Mesos或Standalone）创建毗连
    
  
  
• 功能接口
  
  
  
  • SparkContext是Spark对外的同一接口，为开发者提供访问Spark各种功能的入口
    
  • 具体功能包罗：
    
    
    
    • 创建RDD（弹性分布式数据集）
      
    • 管理累加器（Accumulators）
      
    • 处置惩罚广播变量（Broadcast Variables）
      
    • 设置Spark运行参数
      
    • 作业调治与使命分配
      
    
    
  
  
• 集群毗连
  
  
  
  • 负责创建与Spark集群的毗连
    
  • 管理应用步调与集群资源管理器（如YARN ResourceManager）的通讯
    
  • 示例：在初始化时会指定master URL（如spark://host:port, local等）
    
  
  

范例利用场景

• RDD利用
  
  
  
  • 通过SparkContext可以：
    
    
    
    • 从外部存储体系（如HDFS、S3）创建RDD：sc.textFile("hdfs://path/to/file")
      
    • 并行化聚集：sc.parallelize(Seq(1,2,3))
      
    
    
  
  
• 共享变量管理
  
  
  
  • 累加器（用于聚合信息）：
    
  1. val accum = sc.longAccumulator("My Accumulator")

  复制代码
  
  
  • 广播变量（高效分发大对象）：
    
  1. val broadcastVar = sc.broadcast(Array(1, 2, 3))

  复制代码
• 资源设置
  
  
  
  • 设置应用步调设置：
    
  1. val conf = new SparkConf()
  2.   .setAppName("MyApp")
  3.   .setMaster("local[4]")
  4. val sc = new SparkContext(conf)

  复制代码

注意：在Spark 2.0+版本中，SparkSession已成为新的入口点，但在底层仍会创建SparkContext。对于RDD利用，仍然须要直接利用SparkContext。
从聚集创建RDD

我们在集群的节点上启动 Spark-Shell 举行学习和测试

1. spark-shell --master local[*]

复制代码

假如顺遂启动，你就可以看到如下的画面：

实验运行如下的指令，感受一下

1. Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_412)
2. Type in expressions to have them evaluated.
3. Type :help for more information.
5. scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5))
6. rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24
8. scala> rdd2.getNumPartitions
9. res1: Int = 2
11. scala> rdd2.partitions.length
12. res2: Int = 2
14. scala> val rdd3 = sc.makeRDD(List(1,2,3,4,5))
15. rdd3: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at makeRDD at <console>:24
17. scala> val rdd4 = sc.makeRDD(1 to 100)
18. rdd4: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at makeRDD at <console>:24
20. scala> rdd4.getNumPartitions
21. res3: Int = 2
23. scala>

复制代码

对应的截图如下：

从文件体系创建RDD

用 textFile() 方法来从文件体系中加载数据创建RDD，方法将文件的URI作为参数：

• 本地文件体系
  
• 分布式文件体系 HDFS
  
• Amazon S3的所在
  

1. # 本地系统 注意文件要确保存在
2. val lines = sc.textFile("file:///opt/wzk/1.txt")
3. # 从分布式文件系统加载
4. val lines = sc.textFile("hdfs://h121.wzk.icu:9000/wcinput/wordcount.txt")

复制代码

运行结果如下图所示：

从RDD创建RDD

本质是将一个RDD转换为另一个RDD，从 Transformation
Transformation

RDD的利用算子分为两类：

• Transformation，用来对RDD举行转换，这个利用时耽误实验的（大概是Lazy），Transformation，返回一个新的RDD
  
• Action，用来触发RDD的盘算，得到干系盘算结果大概将结果生存到外部体系中，Action：返回int、double、聚集（不会返回新的RDD）
  

每一个Transformation利用都会产生新的RDD，供给下一个“转换”利用
转换得到RDD是惰性求值，也就是说，整个转换过程只有纪录了转换的轨迹，并不会发生真正的盘算，只有遇到Action利用时，才会发生真正的盘算，开始从学院关系（lineage）源头开始，举行物理的转换利用。

常见转换算子1

map(func)

• 功能：对RDD中的每个元素应用func函数，天生一个包罗转换结果的新RDD
  
• 示例：将整型RDD中的每个元素加1
  1. val rdd = sc.parallelize(List(1, 2, 3))
  2. val mapped = rdd.map(x => x + 1)  // 结果：List(2, 3, 4)

  复制代码
• 特点：输入输出元素逐一对应，不改变数据量
  

filter(func)

• 功能：筛选出使func返回true的元素构成新RDD
  
• 示例：过滤出偶数
  1. val filtered = rdd.filter(x => x % 2 == 0)  // 输入List(1,2,3)，输出List(2)

  复制代码
• 应用场景：数据洗濯、非常值过滤等
  

flatMap(func)

• 功能：每个输入元素可映射为0或多个输出元素（返回一个序列）
  
• 示例：将每行文本拆分为单词
  1. val lines = sc.parallelize(List("hello world", "hi"))
  2. val words = lines.flatMap(_.split(" "))  // 输出：List("hello", "world", "hi")

  复制代码
• 与map区别：map保持1:1映射，flatMap答应1:N映射
  

mapPartitions(func)

• 功能：以分区为单元处置惩罚数据，func吸收一个迭代器（代表整个分区）
  
• 性能上风：恰当须要初始化资源的利用（如数据库毗连）
  1. val partitioned = rdd.mapPartitions(iter => {
  2.   // 每个分区初始化一次数据库连接
  3.   val conn = createConnection()
  4.   iter.map(x => processWithDB(x, conn))
  5. })

  复制代码
• 注意事项：需确保迭代器被完全消耗，否则大概导致资源走漏
  

mapPartitionsWithIndex(func)

• 功能：在mapPartitions底子上增长分区索引参数
  
• 范例应用：调试时检察数据分布
  1. rdd.mapPartitionsWithIndex((index, iter) => {
  2.   println(s"Processing partition $index")
  3.   iter.map(x => x * index)
  4. })

  复制代码
• 参数阐明：func吸收(Int, Iterator[T]) => Iterator[U]，第一个参数是分区索引
  

性能比力

算子调用次数实用场景mapN次（元素级）简朴转换mapPartitionsM次（分区级）须要资源初始化的批量利用flatMapN次数据睁开或过滤注意：mapPartitions系列算子大概引起内存题目，由于须要将整个分区数据加载到内存。
转换算子1测试

map filter

测试如下的代码：

1. val rdd1 = sc.parallelize(1 to 10)
2. val rdd2 = rdd1.map(_*2)
3. val rdd3 = rdd2.filter(_>10)

复制代码

实验结果如下图：

我们可以检察当前的结果，但是当前的利用都是Transformation的，并没有真正的实验。
我们须要通过 collect 触发实验，拿到终极的结果

1. rdd2.collect
2. rdd3.collect

复制代码

将会触发实验，可以看到结果为：

flatMap

我们从HDFS加载一个文件过来

1. val rdd4 = sc.textFile("hdfs://h121.wzk.icu:9000/wcinput/wordcount.txt")
2. rdd4.collect

复制代码

实验结果如下图：

我们利用“a”作为分隔符，对这段内容举行分割：

1. rdd4.flatMap(_.split("a")).collect

复制代码

实验结果如下图：

mapPartitions

1. val rdd5 = rdd1.mapPartitions(iter => iter.map(_*2))

复制代码

实验结果如下

对比 map 和 mapPartitions

上面我们用：

• rdd1.map(_*2)
  
• rdd1.mapPartitions(iter => iter.map(_*2))
  

那么这两种有什么区别呢？

• map：每次只处置惩罚一条数据
  
• mapPartitions：每次处置惩罚一个分区的数据，分区的数据处置惩罚完成后，数据才气开释，资源不敷时容易OOM
  
• 当资源富足时，发起利用 mapPartitions，充实进步处置惩罚服从
  

常见转换算子2

• groupBy（func）：按照传入函数的返回值举行分组，将key类似的值放入一个迭代器
  
• glom（）：将每一个分区形成一个数组，形成新的RDD范例RDD[Array[T]]
  
• sample（withReplacement,fraction,seed）：采样算子，以指定的随机数种子seed随机抽样出数量为fraction的数据，withReplacenent表现抽出数据是否放回，true则放回，false不放回
  
• distinct（[numTasks]）：对RDD元素去重后，返回一个新的RDD，可传入numTasks参数改变RDD分区数
  
• coalesce（numPartitions）：缩减分区数，没有shuffle
  
• repartition（numPartitions）：增长或淘汰分区数，有shuffle
  
• sortBy（func，[ascending], [numTasks]）：利用func对数据举行处置惩罚，对处置惩罚后的结果举行排序
  

宽依靠的算子（shuffle）：groupBy，distinct、repartition、sortBy
转换算子2测试

group by

1. val rdd1 = sc.parallelize(1 to 10)
2. val group = rdd1.groupBy(_%3)
3. group.collect

复制代码

实验的结果如下图：

glom.map

将 RDD 中元素的每10个元素分组

1. val rdd1 = sc.parallelize(1 to 101)
2. val rdd2 = rdd1.glom.map(_.sliding(10, 10).toArray)
3. rdd2.collect

复制代码

实验结果如下图：

sample

对数据采样，fraction表现采样的百分比

1. rdd1.sample(true, 0.2, 2).collect
2. rdd1.sample(false, 0.2, 2).collect
3. rdd1.sample(true, 0.2).collect

复制代码

实验结果如下图：

distinct

对数据举行去重，我们天生一些随机数，然后对这些数值举行去重。

1. val random = scala.util.Random
2. val arr = (1 to 20).map(x => random.nextInt(10))
3. val rdd = sc.makeRDD(arr)
4. rdd.distinct.collect

复制代码

实验结果如下图：

numSlices

对RDD重分区，我们须要多分一些区出来

1. val rdd1 = sc.range(1, 1000, numSlices=10)
2. val rdd2 = rdd1.filter(_%2==0)
3. rdd2.getNumPartitions

复制代码

实验结果如下图：

repartition & coalesce

增长大概淘汰分区

1. rdd2.getNumPartitions
2. # repartition 是增加和缩减分区数
3. val rdd3 = rdd2.repartition(5)
4. # coalesce 是缩减分区数
5. val rdd4 = rdd2.coalesce(5)

复制代码

实验结果如下图：

sortBy

1. rdd.sortBy(x => x).collect
2. rdd.sortBy(x => x).collect

复制代码

实验结果如下：

coalesce & repartition

• repartition：增大大概淘汰分区数，有shuffle
  
• coalesce：一样寻常用于淘汰分区数（此时无shuffle）