4. Spark 数据读取：读取 CSV、JSON、Parquet、JDBC 数据源

在催收业务场景中，数据源格式多样：历史还款记录常以 CSV 存储，客户画像数据多为 JSON 格式，经过 ETL 后的宽表通常以 Parquet 列式存储，而核心账务数据则存放在关系型数据库中。本章将系统讲解 Spark 读取这四种主流数据源的方法，并结合催收业务实例进行演示。

4.1 数据源读取概览

Spark 通过统一的 DataFrameReader 接口读取外部数据，核心 API 为 spark.read.format("...").option("...", "...").load("...")。下表对比了四种数据源的关键特性：

4.2 读取 CSV：催收外呼记录

催收外呼记录通常以 CSV 格式导出，包含字段：案件编号、客户姓名、联系电话、外呼时间、通话时长、催收员编号、催收结果等。以下代码演示如何读取并处理这类数据：

// 读取催收外呼记录 CSV 文件
val callRecords = spark.read
  .format("csv")
  .option("header", "true")                // 首行为列名
  .option("inferSchema", "true")           // 自动推断数据类型
  .option("delimiter", ",")                // 分隔符，默认逗号
  .option("dateFormat", "yyyy-MM-dd HH:mm:ss") // 日期格式
  .option("mode", "PERMISSIVE")            // 容错模式：将损坏行放入 _corrupt_record
  .load("/data/collection/call_records_2024.csv")

// 查看 Schema 和前 5 行
callRecords.printSchema()
callRecords.show(5, truncate = false)

4.3 读取 JSON：客户画像与策略配置

客户画像数据通常以嵌套 JSON 格式存储，包含基本信息、联系人信息、历史行为标签等。催收策略配置也常以 JSON 格式下发。Spark 读取 JSON 时自动处理嵌套结构：

// 读取客户画像 JSON 数据
val customerProfile = spark.read
  .format("json")
  .option("multiLine", "true")             // 支持多行 JSON 对象
  .option("mode", "FAILFAST")              // 遇到格式错误立即失败
  .load("/data/collection/customer_profile/")

// 展开嵌套字段
customerProfile
  .select(
    $"customer_id",
    $"basic_info.name".alias("customer_name"),
    $"basic_info.age",
    $"contact.phone",
    $"tags"                                 // 数组类型
  )
  .show(5, truncate = false)

4.4 读取 Parquet：催收宽表与特征数据

Parquet 是催收数据仓库中最常用的存储格式。经过 ETL 加工后的催收宽表（包含客户基本信息、历史还款行为、催收评分等数百个特征）通常以 Parquet 格式存储。Parquet 自带 Schema 且支持谓词下推，读取效率极高：

// 读取催收宽表 Parquet 文件
val collectionWideTable = spark.read
  .format("parquet")
  .option("mergeSchema", "true")           // 合并多个文件的 Schema
  .load("/data/warehouse/collection_wide_table/")

// 利用谓词下推：只读取逾期天数大于 90 天的记录
val overdue90 = collectionWideTable
  .filter($"overdue_days" > 90)
  .select("case_id", "customer_name", "overdue_days", "collection_score")

overdue90.explain("cost")                  // 查看执行计划，确认下推生效

4.5 读取 JDBC：催收核心数据库

催收系统的核心数据（如案件分配表、还款计划表、催收结果表）通常存储在关系型数据库中。Spark 通过 JDBC 读取时，需要配置连接信息、分区策略和谓词下推：

// 读取 MySQL 催收案件表
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://192.168.1.100:3306/collection_db")
  .option("dbtable", "collection_case")
  .option("user", "spark_user")
  .option("password", "encrypted_password")
  .option("driver", "com.mysql.cj.jdbc.Driver")
  // 分区策略：按案件 ID 范围分片，提升并行度
  .option("partitionColumn", "case_id")
  .option("lowerBound", "1")
  .option("upperBound", "10000000")
  .option("numPartitions", "10")
  // 连接池与 fetch 优化
  .option("fetchSize", "10000")
  .option("connectionPool", "HikariCP")
  .load()

// 使用谓词下推：只读取分配给指定催收团队的案件
val teamCases = jdbcDF
  .filter("team_id = 'T001' AND status = 'ACTIVE'")
  .select("case_id", "customer_name", "assign_date", "priority_level")

teamCases.show(10)

4.6 四种数据源读取流程对比

下图展示了 Spark 读取四种数据源时的核心流程与关键配置项：

4.7 催收业务数据读取实战案例

以下是一个完整的催收数据读取案例，整合了四种数据源，构建催收分析的基础数据集：

// 1. 读取 CSV：催收员每日外呼统计
val dailyCalls = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv("/data/collection/daily_calls/")

// 2. 读取 JSON：客户风险画像
val riskProfile = spark.read
  .option("multiLine", "true")
  .json("/data/collection/risk_profile/")

// 3. 读取 Parquet：催收特征宽表
val features = spark.read
  .parquet("/data/warehouse/collection_features/")

// 4. 读取 JDBC：案件分配信息
val caseAssign = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://db-host:5432/collection")
  .option("dbtable", "case_assignment")
  .option("user", "etl_user")
  .option("password", "secure_pass")
  .option("partitionColumn", "assignment_id")
  .option("lowerBound", "1")
  .option("upperBound", "5000000")
  .option("numPartitions", "8")
  .load()

// 5. 数据关联：构建催收分析宽表
val analysisDF = dailyCalls
  .join(riskProfile, Seq("customer_id"), "left")
  .join(features, Seq("case_id"), "left")
  .join(caseAssign, Seq("case_id"), "left")

analysisDF.createOrReplaceTempView("collection_analysis")

// 6. 业务分析：统计各催收团队的高风险案件处理情况
spark.sql("""
  SELECT
    team_id,
    COUNT(DISTINCT case_id) AS total_cases,
    SUM(CASE WHEN risk_level = 'HIGH' THEN 1 ELSE 0 END) AS high_risk_cases,
    AVG(call_duration) AS avg_call_duration,
    SUM(CASE WHEN result = 'CONTACTED' THEN 1 ELSE 0 END) AS contacted_count
  FROM collection_analysis
  WHERE overdue_days > 30
  GROUP BY team_id
  ORDER BY high_risk_cases DESC
""").show(10)

4.8 常见问题与最佳实践