3. Spark 核心概念：RDD、DataFrame、Dataset、SparkSession、宽依赖与窄依赖

在催收数据处理场景中，理解 Spark 的核心抽象是构建高效 ETL 管道的基础。本章将深入剖析 RDD、DataFrame、Dataset 三大数据抽象，以及驱动任务调度的宽窄依赖机制，并结合催收业务案例说明其应用差异。

3.1 三大数据抽象对比

Spark 提供了三种主要的数据抽象，它们在催收数据处理的灵活性、性能和类型安全方面各有侧重。以下表格从催收业务视角进行对比：

3.2 RDD：弹性分布式数据集

RDD 是 Spark 最基础的数据抽象，具备以下核心特性：

• 不可变性： 对 RDD 的转换操作（如 map、filter）生成新的 RDD，原始数据不变。
• 分区： 数据被切分为多个分区，分布在集群节点上。
• 容错： 通过 Lineage（血统）记录转换过程，分区丢失时可重新计算。
• 惰性求值： 转换操作仅记录依赖关系，遇到 Action 操作（如 count、save）才真正执行。

在催收场景中，RDD 常用于处理非结构化日志或自定义分区逻辑。例如，将催收通话记录按催收员 ID 进行自定义分区，确保同一催收员的记录落在同一分区，减少网络开销：

// 催收通话记录 RDD，按催收员 ID 分区
JavaRDD<CallRecord> callsRDD = sc.textFile("hdfs://.../calls.log")
    .map(line -> parseCallRecord(line))
    .partitionBy(new CollectorPartitioner(10)); // 自定义分区器

3.3 DataFrame：结构化数据处理的基石

DataFrame 在 RDD 之上引入了 Schema 和优化引擎，是催收数据处理的主力工具。它支持 SQL 查询和声明式 API，极大简化了开发。

催收业务示例：统计各催收员当日的有效通话时长总和：

// 使用 DataFrame API
Dataset<Row> callStats = spark
    .read()
    .parquet("hdfs://.../calls/")
    .filter("status = 'CONNECTED' AND duration > 0")
    .groupBy("collector_id")
    .agg(sum("duration").alias("total_duration"))
    .orderBy(col("total_duration").desc());

callStats.show();

3.4 Dataset：类型安全与性能的平衡

Dataset 结合了 RDD 的类型安全和 DataFrame 的优化能力。在催收策略计算中，使用 Dataset 可以在编译时捕获类型错误，同时享受 Tungsten 引擎的高效执行。

催收策略计算示例：计算每个案件的逾期天数并标记高风险案件：

case class Case(caseId: String, dueDate: String, collectorId: String)
case class RiskCase(caseId: String, overdueDays: Int, riskLevel: String)

Dataset<Case> cases = spark.read().parquet("...").as(Encoders.bean(Case.class));
Dataset<RiskCase> riskCases = cases
    .map((MapFunction<Case, RiskCase>) c -> {
        int overdue = calculateOverdueDays(c.dueDate);
        String level = overdue > 90 ? "HIGH" : (overdue > 30 ? "MEDIUM" : "LOW");
        return new RiskCase(c.caseId, overdue, level);
    }, Encoders.bean(RiskCase.class));

riskCases.filter("riskLevel = 'HIGH'").show();

3.5 SparkSession：统一入口

SparkSession 是 Spark 2.0 引入的统一入口点，整合了 SQLContext、HiveContext 和 StreamingContext 的功能。在催收应用中，通过 SparkSession 可以方便地读取多种数据源、执行 SQL 并管理配置。

典型初始化代码：

SparkSession spark = SparkSession
    .builder()
    .appName("CollectionDataPipeline")
    .config("spark.sql.shuffle.partitions", "200") // 控制 Shuffle 分区数
    .config("spark.sql.adaptive.enabled", "true")  // 启用自适应查询执行
    .enableHiveSupport()                           // 如需读取 Hive 表
    .getOrCreate();

3.6 宽依赖与窄依赖

依赖关系决定了 Spark 如何调度任务以及如何处理数据倾斜。理解宽窄依赖对优化催收数据 Shuffle 至关重要。

3.6.1 窄依赖 (Narrow Dependency)

• 定义： 父 RDD 的每个分区最多被子 RDD 的一个分区使用。
• 操作： map、filter、union、mapPartitions。
• 特点： 无需 Shuffle，可在单个节点上流水线执行，容错恢复快。
• 催收场景： 对催收记录进行字段清洗、格式转换。

3.6.2 宽依赖 (Wide Dependency / Shuffle Dependency)

• 定义： 父 RDD 的每个分区可能被子 RDD 的多个分区使用。
• 操作： groupByKey、reduceByKey、join、repartition。
• 特点： 需要 Shuffle，数据在节点间重新分布，开销大，易引发数据倾斜。
• 催收场景： 按催收员分组统计、案件与通话记录关联查询。

以下 SVG 图展示了窄依赖与宽依赖的核心区别：

3.7 催收场景中的依赖优化策略

在催收数据处理中，宽依赖（Shuffle）是性能瓶颈的主要来源。以下策略可有效减少 Shuffle 开销：

1. 使用 reduceByKey 替代 groupByKey： reduceByKey 会在 Map 端进行预聚合，减少 Shuffle 数据量。例如，统计催收员通话次数时，优先使用 reduceByKey 而非 groupByKey。
2. 调整分区数： 通过 spark.sql.shuffle.partitions 控制 Shuffle 后的分区数，避免小文件过多或分区倾斜。
3. 使用广播变量： 当一个大表与一个小表 Join 时，将小表广播到各节点，避免 Shuffle。例如，催收员信息表（小表）与通话记录表（大表）关联时使用 Broadcast Join。
4. 数据预处理： 在写入阶段按催收员 ID 或案件 ID 进行分区，减少后续 Join 的 Shuffle 量。

以下 SVG 图展示了催收数据管道中宽窄依赖的典型应用：

3.8 总结

本章涵盖了 Spark 的核心概念及其在催收数据处理中的应用要点：

• RDD 提供底层控制，适合自定义分区和复杂转换。
• DataFrame 是催收 ETL 的主力，通过 Catalyst 优化器自动提升性能。
• Dataset 在需要类型安全的催收策略计算时使用。
• SparkSession 统一了入口配置，简化了开发。
• 窄依赖 无 Shuffle，适合流水线处理；宽依赖 产生 Shuffle，需重点优化。