4. Spark DataFrame 基础：创建、Schema 定义与常见转换

在组合风险归因场景中，原始数据通常以 CSV、Parquet 或数据库表的形式存在。Spark DataFrame 是处理这类结构化数据的核心抽象，它提供了类似关系型表的视图，并支持分布式计算。本节将系统讲解 DataFrame 的创建方式、Schema 的显式定义以及最常用的转换操作。

4.1 创建 DataFrame 的三种方式

根据数据来源不同，创建 DataFrame 主要有以下三种途径：

方式	适用场景	典型代码
从 RDD 转换	已有 RDD 且需要结构化	spark.createDataFrame(rdd, schema)
从外部数据源读取	文件或数据库	spark.read.csv("path")
从内部数据集合创建	测试或小批量数据	spark.createDataFrame(Seq(...))

4.2 Schema 定义：显式优于隐式

虽然 Spark 支持从数据源自动推断 Schema（inferSchema=true），但在金融风控领域，显式定义 Schema 是更稳健的做法。自动推断可能因数据缺失或格式异常导致类型错误，进而影响后续风险指标计算的准确性。

显式 Schema 定义使用 StructType 和 StructField：

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

val schema = StructType(Array(
  StructField("trade_date", StringType, nullable = false),
  StructField("portfolio_id", StringType, nullable = false),
  StructField("asset_class", StringType, nullable = true),
  StructField("notional", DoubleType, nullable = false),
  StructField("delta", DoubleType, nullable = true),
  StructField("gamma", DoubleType, nullable = true)
))

val df = spark.read
  .schema(schema)
  .option("header", "true")
  .csv("hdfs://data/risk/positions.csv")

4.3 常见转换操作

DataFrame 的转换操作遵循 惰性求值 原则，即转换只构建执行计划，不触发实际计算。以下四个操作是风险归因数据处理中最常用的：

4.3.1 select：列选择与投影

select 用于从 DataFrame 中选取特定列，支持列名、表达式和别名：

// 选取基础字段
df.select("trade_date", "portfolio_id", "notional")

// 使用表达式计算名义本金的对数
df.select(
  col("trade_date"),
  col("notional"),
  log("notional").alias("log_notional")
)

4.3.2 filter：行过滤

filter（或 where）用于按条件筛选行，在风控中常用于隔离特定资产类别或风险阈值：

// 筛选权益类资产且名义本金大于 1,000,000
df.filter(col("asset_class") === "Equity" && col("notional") > 1000000)

// 筛选 delta 为空的记录（数据质量问题）
df.filter(col("delta").isNull)

4.3.3 withColumn：新增或替换列

withColumn 是风险归因中最灵活的操作，用于衍生风险指标、标记分类或转换数据类型：

// 新增列：计算 delta 调整后的名义本金
df.withColumn("delta_adjusted_notional", col("notional") * col("delta"))

// 新增列：根据资产类别打标签
df.withColumn("risk_group",
  when(col("asset_class") === "Equity", "权益风险")
  .when(col("asset_class") === "FixedIncome", "利率风险")
  .otherwise("其他风险")
)

// 替换列：将 trade_date 转为日期类型
df.withColumn("trade_date", to_date(col("trade_date"), "yyyy-MM-dd"))

4.4 转换操作对比与选择

操作	功能	返回类型	典型风控用途
select	列投影、表达式计算	DataFrame	提取风险因子列、计算衍生指标
filter	行过滤	DataFrame	隔离特定资产类别、剔除异常值
withColumn	新增/替换单列	DataFrame	添加风险标签、转换数据类型
drop	删除列	DataFrame	移除中间计算列、精简 Schema

4.5 实战：组合风险数据预处理流水线

将上述操作串联起来，构建一个典型的组合风险数据预处理流水线：

// 1. 定义 Schema
val riskSchema = StructType(Array(
  StructField("date", StringType, nullable = false),
  StructField("port_id", StringType, nullable = false),
  StructField("asset", StringType, nullable = true),
  StructField("amount", DoubleType, nullable = false),
  StructField("delta", DoubleType, nullable = true),
  StructField("gamma", DoubleType, nullable = true)
))

// 2. 读取数据
val rawDF = spark.read
  .schema(riskSchema)
  .csv("hdfs://data/risk/daily_positions.csv")

// 3. 转换流水线
val processedDF = rawDF
  .filter(col("amount") > 0)                              // 剔除零名义本金
  .withColumn("date", to_date(col("date"), "yyyyMMdd"))   // 日期标准化
  .withColumn("risk_level",                               // 风险等级标记
    when(col("delta") > 0.5, "高敏感")
    .when(col("delta") > 0.2, "中敏感")
    .otherwise("低敏感"))
  .select("date", "port_id", "asset", "amount", "risk_level", "delta", "gamma")

processedDF.show(10)