2、Spark环境准备：Spark安装与配置、SparkSession初始化、数据读取基础

在开始特征工程样本偏差校验之前，必须搭建一个稳定、可复现的Spark运行环境。本章将指导你完成从Spark安装到数据读取的全流程，确保后续所有校验逻辑都能在统一的计算框架上执行。

2.1 Spark安装与配置

Spark支持多种部署模式：本地模式（Local）、独立集群模式（Standalone）、YARN模式以及Kubernetes模式。对于特征工程样本偏差校验的开发与测试阶段，推荐使用本地模式，它无需额外集群资源，即可快速验证代码逻辑。

2.1.1 下载与安装步骤

1. 下载Spark：访问 https://spark.apache.org/downloads.html，选择预编译好的Hadoop版本（如 spark-3.4.1-bin-hadoop3.tgz）。
2. 解压安装：将压缩包解压至指定目录，例如 /opt/spark 或 C:\spark。
3. 配置环境变量：设置 SPARK_HOME 并将 $SPARK_HOME/bin 加入 PATH。
4. 验证安装：在终端执行 spark-shell --version，确认版本信息正确输出。

# Linux/macOS 环境变量配置示例
export SPARK_HOME=/opt/spark
export PATH=$SPARK_HOME/bin:$PATH

# 验证安装
spark-shell --version

2.2 SparkSession初始化

SparkSession是Spark 2.0+的统一入口点，它整合了SQLContext、HiveContext和StreamingContext的功能。在样本偏差校验任务中，我们通过SparkSession来配置资源、读取数据源以及执行DataFrame操作。

2.2.1 标准初始化模板

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("SampleBiasValidation")
  .master("local[*]")  // 本地模式，使用所有可用核心
  .config("spark.sql.shuffle.partitions", "10") // 小数据量时减少分区数
  .config("spark.sql.adaptive.enabled", "true") // 启用自适应查询执行
  .getOrCreate()

// 设置日志级别，减少控制台干扰
spark.sparkContext.setLogLevel("WARN")

2.2.2 关键配置参数说明

2.3 数据读取基础

样本偏差校验的第一步是加载数据。Spark支持从多种数据源读取数据，包括CSV、Parquet、JSON、Avro以及JDBC等。在特征工程场景中，最常用的格式是Parquet（列式存储、压缩率高）和CSV（通用性强、易于查看）。

2.3.1 读取CSV文件

// 读取CSV文件，自动推断Schema
val dfCsv = spark.read
  .option("header", "true")          // 第一行为列名
  .option("inferSchema", "true")     // 自动推断数据类型
  .option("delimiter", ",")          // 分隔符
  .option("mode", "PERMISSIVE")      // 容错模式，损坏行置为null
  .csv("path/to/sample_data.csv")

dfCsv.printSchema()
dfCsv.show(5, truncate = false)

2.3.2 读取Parquet文件

// 读取Parquet文件（自带Schema，无需推断）
val dfParquet = spark.read
  .option("mergeSchema", "true")     // 合并多个Parquet文件的Schema
  .parquet("path/to/feature_data/")

dfParquet.printSchema()
println(s"总记录数：${dfParquet.count()}")

2.3.3 读取JSON文件

// 读取多行JSON文件
val dfJson = spark.read
  .option("multiLine", "true")       // 一个JSON对象跨多行
  .json("path/to/nested_data.json")

// 展平嵌套结构
import org.apache.spark.sql.functions._
val dfFlattened = dfJson.select(
  col("id"),
  col("features.*")                  // 展开features结构体
)

2.4 数据读取最佳实践

• Schema预定义：对于生产环境，建议手动定义Schema而非使用 inferSchema，避免每次读取都触发额外的数据扫描。
• 分区读取：如果数据按日期或地域分区，使用 spark.read.parquet("base_path/") 自动发现分区列。
• 采样与缓存：在偏差校验初期，使用 .sample(withReplacement = false, fraction = 0.1) 进行数据采样，配合 .cache() 加速迭代分析。
• 空值处理：读取时设置 .option("nullValue", "NA") 或 .option("nanValue", "NaN")，统一空值表示。

// 手动定义Schema示例
import org.apache.spark.sql.types._

val schema = StructType(Array(
  StructField("user_id", StringType, nullable = false),
  StructField("age", IntegerType, nullable = true),
  StructField("income", DoubleType, nullable = true),
  StructField("label", IntegerType, nullable = false)
))

val df = spark.read
  .schema(schema)
  .option("header", "true")
  .csv("path/to/users.csv")