3. 数据读取与基础ETL：使用Spark读取Hive/Parquet/CSV数据、数据清洗、缺失值处理、数据类型转换

在风控特征样本分层中，数据读取与基础ETL是构建高质量样本集的第一步。本章将系统讲解如何使用Spark从Hive、Parquet、CSV三种常见存储介质中读取数据，并完成数据清洗、缺失值处理与数据类型转换等核心操作。掌握这些基础能力，是后续进行特征工程与样本分层的前提。

3.1 数据读取：三种存储介质的Spark接入方式

风控数据通常以Hive表（离线数仓）、Parquet文件（列式存储，高压缩比）或CSV文件（外部数据源）的形式存在。Spark提供了统一的DataFrame API来读取这些数据源，但各自有特定的配置参数与优化技巧。

数据源 读取方式 核心参数 适用场景
Hive spark.sql("SELECT * FROM db.table") spark.sql.warehouse.dir、hive.metastore.uris 离线数仓表，大规模结构化数据
Parquet spark.read.parquet("path") mergeSchema、compression 列式存储，高压缩比，Spark原生优化
CSV spark.read.option("header","true").csv("path") delimiter、inferSchema、mode 外部数据源，非结构化或半结构化数据

关键提示： 在风控场景中，Hive表通常存储用户画像、交易流水等核心数据；Parquet文件常用于中间特征存储；CSV则多用于外部黑名单、第三方数据接入。建议优先使用Parquet格式进行中间结果存储，其列式存储特性在后续特征筛选时能显著提升I/O效率。

3.2 数据清洗：处理脏数据与异常值

风控原始数据中常包含重复记录、格式错误、逻辑异常（如年龄为负数、交易金额为0但状态为成功）等问题。数据清洗的目标是识别并处理这些脏数据，确保样本集的可靠性。

3.2.1 常见脏数据类型与处理策略

  • 重复数据： 使用dropDuplicates()基于主键或关键字段去重。
  • 格式错误： 使用when() + cast()进行条件转换，或使用regexp_replace()修正字符串格式。
  • 逻辑异常： 定义业务规则（如年龄在0~120之间），使用filter()或where()剔除异常记录。
// 数据清洗示例：去重 + 格式修正 + 逻辑过滤
val cleanDF = rawDF
  .dropDuplicates("user_id", "event_time")          // 基于用户ID和事件时间去重
  .withColumn("age", 
    when($"age".between(0, 120), $"age")            // 年龄逻辑校验
    .otherwise(lit(null)))
  .withColumn("phone", 
    regexp_replace($"phone", "[^0-9]", ""))         // 手机号去除非数字字符
  .filter($"amount" > 0 || $"status" === "success") // 交易金额与状态逻辑一致

实操建议： 在风控特征样本分层中，建议将清洗逻辑封装为独立的UDF或函数，便于在多个样本集间复用。同时，清洗过程中应记录被剔除的数据量，用于后续数据质量监控。

3.3 缺失值处理：填充、删除与标记

缺失值是风控特征中的常见问题，处理不当会导致模型偏差或样本量不足。Spark提供了灵活的缺失值处理API，支持按列、按统计量或按业务规则进行填充。

3.3.1 缺失值处理策略对比

策略 方法 适用场景 注意事项
删除 na.drop() 缺失比例低（<5%）且随机缺失 可能导致样本量骤减，需评估影响
均值/中位数填充 na.fill() + agg(mean) 连续型特征，分布较对称 会降低特征方差，影响模型区分度
众数填充 na.fill() + agg(mode) 离散型特征，如性别、职业 需自定义mode计算逻辑
业务默认值填充 na.fill(value, cols) 有明确业务含义的字段 如年龄缺失填充-1，收入缺失填充0
标记缺失 when(isnull($"col"), 1).otherwise(0) 缺失本身具有信息量 增加一列缺失标记，保留原始缺失 
// 缺失值处理综合示例
val filledDF = cleanDF
  // 1. 删除关键字段缺失的记录
  .na.drop(Seq("user_id", "event_time"))
  // 2. 连续型特征：均值填充
  .withColumn("income", 
    when($"income".isNull, avgIncome).otherwise($"income"))
  // 3. 离散型特征：业务默认值填充
  .na.fill(Map("gender" -> "unknown", "occupation" -> "other"))
  // 4. 标记缺失：增加缺失指示列
  .withColumn("income_missing", when($"income".isNull, 1).otherwise(0))

风险提示： 在风控样本分层中，不同分层的缺失值模式可能不同（如高逾期人群的某些特征缺失率更高）。建议在分层后分别统计缺失率，避免全局填充掩盖分层间的差异。

3.4 数据类型转换：确保特征计算的正确性

Spark在读取数据时，尤其是CSV文件，常出现类型推断错误（如将数值型读为字符串）。数据类型转换是确保后续特征计算、模型训练正确性的基础。

3.4.1 常见类型转换场景

  • 字符串转数值： cast("double")、cast("int")，注意处理非数字字符。
  • 字符串转日期： to_date()、to_timestamp()，需指定日期格式。
  • 数值转类别： 使用when()进行分箱，或使用bucketizer进行等频/等距分箱。
  • 布尔类型： 使用cast("boolean")，或when()将0/1转为true/false。
// 数据类型转换示例
val typedDF = filledDF
  // 1. 字符串转数值：处理千分位逗号
  .withColumn("amount", 
    regexp_replace($"amount", ",", "").cast("double"))
  // 2. 字符串转日期：指定格式
  .withColumn("event_date", to_date($"event_date_str", "yyyyMMdd"))
  // 3. 数值转类别：年龄分箱
  .withColumn("age_group", 
    when($"age" < 18, "teen")
    .when($"age" < 35, "young")
    .when($"age" < 60, "middle")
    .otherwise("elder"))
  // 4. 布尔类型转换
  .withColumn("is_risk", $"risk_flag".cast("boolean"))

性能优化： 在大型数据集上，cast()操作会触发全表扫描。建议在读取阶段就通过schema参数指定数据类型，避免后续大规模转换。对于CSV文件，使用inferSchema时需注意其性能开销，建议手动指定schema。

3.5 数据读取与基础ETL流程总览

下图展示了从数据读取到基础ETL的完整流程，包括三种数据源的接入、清洗、缺失值处理与类型转换，最终输出可用于特征工程的标准样本集。

数据读取与基础ETL流程 数据读取与基础ETL流程 Hive表 Parquet文件 CSV文件 spark.sql() spark.read.parquet() spark.read.csv() 数据清洗 缺失值处理 数据类型转换 标准样本集（DataFrame）

3.6 完整代码示例：从Hive读取到基础ETL

以下代码展示了从Hive表读取风控数据，并完成清洗、缺失值处理与类型转换的完整流程。该代码可直接应用于风控特征样本分层的前置处理阶段。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

// 1. 初始化SparkSession，启用Hive支持
val spark = SparkSession.builder()
  .appName("RiskControl_ETL")
  .config("spark.sql.warehouse.dir", "/user/hive/warehouse")
  .enableHiveSupport()
  .getOrCreate()

// 2. 从Hive表读取数据
val rawDF = spark.sql("""
  SELECT user_id, event_time, age, gender, income, amount, 
         risk_flag, phone, occupation
  FROM risk_control.user_behavior
  WHERE dt = '2024-01-15'
""")

// 3. 数据清洗
val cleanDF = rawDF
  .dropDuplicates("user_id", "event_time")
  .withColumn("age", 
    when($"age".between(0, 120), $"age").otherwise(lit(null)))
  .withColumn("phone", regexp_replace($"phone", "[^0-9]", ""))
  .filter($"amount" > 0 || $"risk_flag" === 1)

// 4. 缺失值处理
val avgIncome = cleanDF.select(avg("income")).first().getDouble(0)
val filledDF = cleanDF
  .na.drop(Seq("user_id", "event_time"))
  .withColumn("income", when($"income".isNull, avgIncome).otherwise($"income"))
  .na.fill(Map("gender" -> "unknown", "occupation" -> "other"))
  .withColumn("income_missing", when($"income".isNull, 1).otherwise(0))

// 5. 数据类型转换
val typedDF = filledDF
  .withColumn("amount", $"amount".cast("double"))
  .withColumn("event_date", to_date($"event_time", "yyyy-MM-dd HH:mm:ss"))
  .withColumn("age_group", 
    when($"age" < 18, "teen")
    .when($"age" < 35, "young")
    .when($"age" < 60, "middle")
    .otherwise("elder"))
  .withColumn("is_risk", $"risk_flag".cast("boolean"))

// 6. 输出结果
typedDF.show(10, false)
typedDF.printSchema()

扩展思考： 在实际风控项目中，基础ETL通常需要处理TB级数据。建议使用Spark的cache()或checkpoint()在关键节点持久化中间结果，避免重复计算。同时，利用explain()查看执行计划，优化数据倾斜问题。

3.7 本章小结

本章系统讲解了Spark在风控特征样本分层中的数据读取与基础ETL操作。通过Hive、Parquet、CSV三种数据源的接入方式，结合数据清洗、缺失值处理与数据类型转换的实战技巧，为后续特征工程与样本分层奠定了坚实的数据基础。掌握这些基础能力，能够确保风控样本集的质量与一致性，提升模型训练的稳定性和预测效果。

← 上一章 📖 返回目录 下一章 →