4. Spark SQL 环境搭建：本地模式与集群模式配置、SparkSession 初始化、读取金融数据源（CSV/Parquet/Hive）

金融特征巡检的第一步是搭建稳定、可复用的 Spark SQL 运行环境。本章将围绕 本地模式（开发调试）与 集群模式（生产调度）的配置差异、SparkSession 的标准化初始化流程，以及金融场景下最常见的三种数据源（CSV、Parquet、Hive）的读取方式展开。掌握这些基础，后续的特征缺失巡检逻辑才能高效落地。

4.1 本地模式与集群模式配置对比

Spark 支持多种运行模式，在金融特征工程中，本地模式 用于单机数据探索与代码调试，YARN 集群模式 用于生产环境的大规模调度。下表列出了两种模式在关键配置上的差异：

4.2 SparkSession 标准化初始化

Spark 2.0 之后，SparkSession 是统一的入口点，替代了旧的 SparkContext、SQLContext 和 HiveContext。在金融特征巡检项目中，建议将初始化逻辑封装为工具方法，确保环境一致且可配置。

以下是一个同时支持本地模式与 YARN 模式的初始化模板：

import org.apache.spark.sql.SparkSession

object SparkEnvBuilder {

  def createSparkSession(appName: String, master: String = "local[*]"): SparkSession = {
    val builder = SparkSession.builder()
      .appName(appName)
      .master(master)
      .config("spark.sql.shuffle.partitions", 
        if (master.startsWith("local")) "8" else "200")
      .config("spark.sql.adaptive.enabled", "true")
      .config("spark.sql.adaptive.coalescePartitions.enabled", "true")

    // 金融场景：开启 Hive 支持（若需读取 Hive 表）
    if (master != "local[*]") {
      builder
        .config("hive.metastore.uris", "thrift://hive-metastore:9083")
        .enableHiveSupport()
    }

    builder.getOrCreate()
  }
}

4.3 读取金融数据源：CSV / Parquet / Hive

金融特征数据通常以 CSV（交易流水）、Parquet（列式存储特征宽表）、Hive（数仓分层表）三种形式存储。Spark SQL 提供了统一的 DataFrameReader API，但不同格式的配置细节差异较大。

4.3.1 读取 CSV（交易流水 / 对账文件）

CSV 是金融系统间交换数据的常见格式，但常存在字段缺失、引号嵌套、日期格式不统一等问题。以下配置可提升解析鲁棒性：

val tradeDF = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .option("delimiter", ",")
  .option("quote", "\"")
  .option("escape", "\"")
  .option("mode", "PERMISSIVE")   // 容错模式，坏行置 null
  .option("dateFormat", "yyyy-MM-dd")
  .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
  .csv("hdfs://nameservice1/data/finance/trade/2025/04/01/")

4.3.2 读取 Parquet（特征宽表 / 中间层）

Parquet 是金融特征存储的首选格式，支持列式压缩、谓词下推和 Schema 演进。读取方式最为简洁：

val featureDF = spark.read
  .option("mergeSchema", "true")   // 允许字段新增或类型兼容变化
  .parquet("hdfs://nameservice1/data/finance/features/daily/")

Parquet 读取时，Spark SQL 会自动利用 统计信息（min/max 值）进行文件裁剪，在特征缺失巡检中可大幅减少扫描量。例如，巡检某只股票近 30 天的特征缺失率时，仅读取对应分区的 Parquet 文件。

4.3.3 读取 Hive 表（数仓分层表）

在大型金融机构中，特征数据通常已入 Hive 数仓。读取 Hive 表需要先启用 Hive 支持，然后直接使用 SQL 或 table API：

// 方式一：使用 SQL
val riskDF = spark.sql("SELECT * FROM fin_dw.risk_feature WHERE dt = '2025-04-01'")

// 方式二：使用 table API
val assetDF = spark.table("fin_dw.asset_feature")
  .filter($"dt" === "2025-04-01")
  .select("customer_id", "asset_level", "missing_ratio")

4.4 三种数据源读取流程对比

下面的 SVG 图展示了从数据源到 Spark DataFrame 的标准化读取流程，以及各环节的关键配置点：

4.5 完整初始化与数据读取示例

将上述知识点整合为一个完整的 Scala 脚本，展示从 SparkSession 创建到多数据源读取的全过程：

import org.apache.spark.sql.SparkSession

object FinanceDataLoader {

  def main(args: Array[String]): Unit = {
    // 1. 初始化 SparkSession（本地模式）
    val spark = SparkSession.builder()
      .appName("FinanceFeatureCheck")
      .master("local[4]")
      .config("spark.sql.shuffle.partitions", "8")
      .config("spark.sql.adaptive.enabled", "true")
      .getOrCreate()

    import spark.implicits._

    // 2. 读取 CSV 交易流水
    val tradeDF = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .option("mode", "PERMISSIVE")
      .csv("data/finance/trade_sample.csv")

    // 3. 读取 Parquet 特征宽表
    val featureDF = spark.read
      .option("mergeSchema", "true")
      .parquet("data/finance/features/")

    // 4. 读取 Hive 表（需 enableHiveSupport）
    // val hiveDF = spark.sql("SELECT * FROM fin_dw.risk_feature WHERE dt='2025-04-01'")

    // 5. 打印 Schema 验证
    tradeDF.printSchema()
    featureDF.printSchema()

    println("所有数据源加载完成，准备进行特征缺失巡检。")
    spark.stop()
  }
}

4.6 环境搭建常见问题与排查

下表总结了金融数据工程师在环境搭建阶段最常遇到的三个问题及解决方案：