3. 原始数据接入：读取CSV/Parquet/JSON格式数据、Schema推断与自定义、数据分区策略

信用评分建模的第一步是将分散在数据仓库、日志系统或第三方数据源中的原始数据加载到Spark DataFrame中。本章聚焦于三种最常见的文件格式（CSV、Parquet、JSON）的读取方式，Schema的自动推断与手动定义，以及如何通过合理的数据分区策略提升后续处理效率。

3.1 三种主流格式的读取方式

Spark SQL提供了统一的DataFrameReader接口，通过.format()指定格式，配合.load()或专用方法（如.csv()）完成数据接入。下表对比了三种格式在信用评分场景下的核心差异：

格式	典型场景	压缩支持	Schema携带	读取速度	存储效率
CSV	业务导出的原始申请表、交易流水	gzip / bzip2	无（需推断或指定）	中等	低（文本冗余）
Parquet	ETL中间层、特征存储、模型训练集	Snappy / ZSTD	有（自描述）	快（列式+谓词下推）	高（列式+编码）
JSON	埋点日志、API响应、半结构化数据	gzip	有（嵌套结构）	较慢（文本解析）	低

3.1.1 CSV读取实战

CSV是信用评分中最常见的原始数据格式，但也是最容易出错的。以下代码展示了带典型配置项的读取方式：

// 读取带表头的CSV，自动推断数据类型，处理空值和日期
val rawDF = spark.read
  .format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .option("delimiter", ",")
  .option("quote", "\"")
  .option("escape", "\\")
  .option("mode", "PERMISSIVE")   // 容错模式：将坏行放入_corrupt_record
  .option("dateFormat", "yyyy-MM-dd")
  .option("nullValue", "NULL")
  .option("pathGlobFilter", "*.csv")
  .load("/data/credit/raw/applications/")

// 查看推断出的Schema
rawDF.printSchema()

3.1.2 Parquet读取实战

Parquet是Spark生态中的一等公民，在特征工程中间环节强烈推荐使用。读取方式最为简洁：

// Parquet自带Schema，无需额外配置
val featureDF = spark.read
  .format("parquet")
  .load("/data/credit/features/derived/")

// 利用Parquet的谓词下推能力，只读取需要的列和分区
val filteredDF = spark.read
  .format("parquet")
  .option("basePath", "/data/credit/features/")
  .load("/data/credit/features/derived/dt=2025-01-01/")
  .select("user_id", "credit_score", "income_level")

3.1.3 JSON读取实战

当信用评分需要引入外部征信报告或设备指纹等半结构化数据时，JSON格式非常常见：

// 读取多行JSON（每行一个完整JSON对象）
val jsonDF = spark.read
  .format("json")
  .option("multiLine", "false")
  .option("mode", "FAILFAST")   // 遇到格式错误立即失败，保证数据质量
  .option("primitivesAsString", "false")
  .load("/data/credit/external/device_fingerprint/")

// 展开嵌套结构
val flatDF = jsonDF
  .withColumn("device_os", $"device_info.os")
  .withColumn("screen_resolution", $"device_info.screen")

3.2 Schema推断与自定义

Schema定义了DataFrame的列名、数据类型和可空性。在信用评分中，Schema的准确性直接影响后续特征计算的正确性。

3.2.1 自动推断的优缺点

维度	优点	缺点
开发效率	无需手动定义，快速预览数据	推断需要额外扫描数据（1-2轮），大文件时耗时
类型精度	自动识别整数、浮点、日期等	可能误判（如身份证号被推断为IntegerType）
容错性	对缺失值自动设为nullable	无法处理自定义格式（如“1,234.56”带千分符的数字）

3.2.2 自定义Schema的最佳实践

在生产级信用评分流水线中，强烈建议使用自定义Schema，原因有三：

• 性能提升：跳过推断扫描，读取速度提升30%-50%。
• 类型安全：避免“年龄”被推断为StringType，“收入”因空值被推断为StringType。
• 数据质量：Schema中的类型约束会在读取时自动进行类型校验。

import org.apache.spark.sql.types._

val customSchema = StructType(Array(
  StructField("user_id", StringType, nullable = false),
  StructField("application_date", DateType, nullable = true),
  StructField("age", IntegerType, nullable = true),
  StructField("annual_income", DoubleType, nullable = true),
  StructField("credit_card_debt", DecimalType(12, 2), nullable = true),
  StructField("loan_purpose", StringType, nullable = true)
))

val preciseDF = spark.read
  .format("csv")
  .option("header", "true")
  .schema(customSchema)   // 使用自定义Schema，跳过推断
  .load("/data/credit/raw/applications/")

3.3 数据分区策略

分区是Spark管理大数据集的核心手段，直接影响读取效率、Shuffle开销和并行度。在信用评分场景中，分区策略需要兼顾数据量级和查询模式。

3.3.1 文件级分区 vs. DataFrame分区

分区类型	控制方式	作用范围	典型用途
目录分区	Hive风格目录（如dt=2025-01-01/）	文件系统层面	按日期、地区、产品线划分原始数据
DataFrame分区	.repartition() / .coalesce()	内存中RDD/DataFrame	控制Shuffle并行度、优化Join

3.3.2 目录分区读取示例

信用评分数据通常按日期分区存储，读取时利用分区裁剪只加载必要数据：

// 读取2025年1月全量数据（自动发现dt分区列）
val janDF = spark.read
  .format("parquet")
  .load("/data/credit/raw/transactions/dt=2025-01-*/")

// 显式指定分区过滤（更高效）
val specificDF = spark.read
  .format("parquet")
  .load("/data/credit/raw/transactions/")
  .where($"dt" >= "2025-01-01" && $"dt" <= "2025-01-31")

3.3.3 分区数调优原则

• 目标分区大小：每个分区128MB-256MB（对应HDFS块大小）。
• 并行度公式：分区数 ≈ 总数据量 / 目标分区大小 × 集群核心数系数（通常1-3倍）。
• 避免小文件问题：信用评分中每日增量数据可能只有几MB，应合并为小时级或天级分区。

// 读取后重新分区，适配后续Join操作
val repartitionedDF = rawDF
  .repartition(200, $"user_id")   // 按用户ID哈希分区，保证相同用户的数据在同一分区
  .cache()

3.4 本章小结

• 格式选择：原始数据用CSV（带自定义Schema），中间特征用Parquet（列式+压缩），外部数据用JSON（嵌套结构）。
• Schema策略：开发阶段用inferSchema快速探索，生产阶段必须使用自定义Schema保证类型安全和性能。
• 分区核心：目录分区用于数据隔离和裁剪，DataFrame分区用于控制并行度和Shuffle。信用评分场景优先按user_id和dt组合分区。