4. RDD 基础：概念、创建、分区与算子分类
RDD（Resilient Distributed Dataset，弹性分布式数据集）是 Spark 最核心的抽象概念，也是催收触达数据处理流水线的底层基石。理解 RDD 的设计思想与操作方式，是掌握 Spark 性能调优与任务编排的前提。
4.1 RDD 核心概念
RDD 是一个不可变、可分区、可并行计算的数据集合。它具备以下五个关键特性：
| 特性 | 说明 | 催收场景示例 |
|---|---|---|
| 分区（Partition） | 数据被切分为多个分片，每个分区对应一个计算任务 | 1000 万条催收记录按客户 ID 哈希分到 200 个分区 |
| 不可变性（Immutable） | RDD 一旦创建就不能修改，只能通过算子生成新 RDD | 过滤逾期天数 > 30 天的记录，生成新的 RDD |
| 弹性（Resilient） | 分区丢失后可依据血缘关系（Lineage）重新计算 | 某节点宕机，Spark 自动重算丢失的分区 |
| 惰性求值（Lazy Evaluation） | 转换操作只记录血缘，遇到行动操作才真正计算 | 多次 filter 不触发计算，直到 count() 才执行 |
| 可缓存（Cacheable） | 可将中间结果存入内存或磁盘，加速重复查询 | 将催收策略匹配结果 cache()，供多个报表复用 |
💡 催收经验： 在催收数据清洗阶段，RDD 的不可变性天然支持“数据血缘审计”——每一笔记录从原始数据到最终标签的转换链路都可以通过 RDD 的依赖关系追溯，满足合规审计要求。
4.2 RDD 创建方式
在催收数据处理中，RDD 的创建主要有三种途径：
- 从外部存储系统加载：读取 HDFS、HBase、本地文件等。
- 从 Scala/Java 集合创建：通过
parallelize方法将内存集合转为 RDD。 - 从已有 RDD 转换：通过
map、filter等算子生成新 RDD。
📌 催收最佳实践： 生产环境优先使用 textFile 或 wholeTextFiles 读取催收记录文件，避免使用 parallelize 加载大数据集，否则会导致 Driver 内存溢出。
// 方式一：从 HDFS 加载催收记录
val rdd1 = sc.textFile("hdfs://namenode:8020/collection/records/2025-04/*.csv")
// 方式二：从集合创建（仅用于小数据测试）
val testData = Seq(("张三", 35, "M1"), ("李四", 60, "M2"))
val rdd2 = sc.parallelize(testData, numSlices = 4)
// 方式三：从已有 RDD 转换
val overdueRDD = rdd1.filter(line => line.split(",")(2).toInt > 30)
4.3 RDD 分区机制
分区是 RDD 并行计算的基本单位。每个分区对应一个 Task，分区数决定了集群的并行度。
| 分区方式 | 说明 | 催收适用场景 |
|---|---|---|
| 默认分区 | textFile 默认分区数 = 文件 Block 数（HDFS 128MB/块） | 原始催收日志按块分布，无需额外调整 |
| 手动指定 | 通过 sc.textFile(path, minPartitions) 指定最小分区数 |
小文件合并场景，减少 Task 调度开销 |
| 哈希分区 | 通过 partitionBy(new HashPartitioner(n)) 按 Key 哈希 |
按客户 ID 分区，避免 Shuffle 时数据倾斜 |
| 范围分区 | 通过 partitionBy(new RangePartitioner(n, rdd)) 按 Key 范围 |
按逾期天数范围分区，便于区间统计 |
| 自定义分区 | 继承 Partitioner 实现 getPartition 方法 |
按催收员 ID 分区，实现数据本地性 |
⚠️ 分区陷阱： 分区数并非越多越好。分区过多会导致 Task 调度开销剧增，每个 Task 处理的数据量过小，反而降低吞吐量。催收场景建议分区数 = Executor 核心数 × 2~3 倍。
4.4 RDD 算子分类
RDD 算子分为两大类：转换算子（Transformation） 和 行动算子（Action）。这是 Spark 惰性求值机制的核心体现。
4.4.1 转换算子（Transformation）
转换算子将一个 RDD 转换为另一个 RDD，属于惰性操作，只记录血缘关系，不触发实际计算。
| 算子 | 功能 | 催收应用示例 |
|---|---|---|
map(func) |
对每个元素应用函数，1→1 映射 | 将 CSV 行解析为 Case Class |
filter(func) |
保留满足条件的元素 | 过滤出逾期天数 > 90 天的客户 |
flatMap(func) |
对每个元素应用函数，1→N 映射 | 将一条催收记录拆成多条联系人记录 |
groupByKey() |
按 Key 分组，返回 (K, Iterable[V]) | 按催收员分组统计待办任务 |
reduceByKey(func) |
按 Key 聚合，先本地聚合再全局 | 统计每个客户的累计逾期金额 |
join(otherRDD) |
两个 RDD 按 Key 内连接 | 催收记录关联客户画像数据 |
distinct() |
去重 | 去重当天重复触达的记录 |
4.4.2 行动算子（Action）
行动算子触发实际计算，将结果返回 Driver 或写入外部存储。
| 算子 | 功能 | 催收应用示例 |
|---|---|---|
collect() |
将 RDD 所有元素收集到 Driver | 小数据量调试时查看前几条记录 |
count() |
返回 RDD 中元素总数 | 统计今日待催收客户总数 |
take(n) |
返回前 n 个元素 | 抽样查看催收记录格式 |
reduce(func) |
并行聚合所有元素 | 计算所有催收员的总接通次数 |
saveAsTextFile(path) |
将 RDD 写入文本文件 | 将清洗后的催收结果写入 HDFS |
foreach(func) |
对每个元素执行副作用操作 | 将催收结果写入外部数据库 |
4.5 催收场景算子选择建议
🔑 性能关键： 在催收数据处理中，优先使用
reduceByKey 而非 groupByKey。前者会在 Map 端做预聚合，大幅减少 Shuffle 数据量。例如统计每个催收员的接通率时，reduceByKey 比 groupByKey 快 3~5 倍。
// ❌ 不推荐：groupByKey 导致全量 Shuffle
val result1 = rdd.groupByKey().mapValues(iter => iter.sum)
// ✅ 推荐：reduceByKey 先本地聚合再 Shuffle
val result2 = rdd.reduceByKey(_ + _)