4. 使用Spark进行数据探索与特征工程

在 Databricks 平台上，Spark 是数据探索与特征工程的核心执行引擎。无论数据存放在 Delta Lake、Parquet 还是外部存储中，我们都可以通过 DataFrame API 或 Spark SQL 对数据进行清洗、聚合、变换和编码，最终生成适合机器学习模型训练的特征表。本节围绕 DataFrame API、SQL、窗口函数、聚合、缺失值处理、特征编码与缩放这几项关键能力，说明它们在实际流程中的位置与用法。

DataFrame API 与 SQL：两种等价入口

DataFrame API 以编程方式表达变换，适合在 Python 中做链式处理；Spark SQL 则以声明式语句描述同一套逻辑。二者最终都会经过 Catalyst 优化器生成执行计划，因此性能上没有本质差异，选择取决于可读性与团队习惯。在 Databricks 中，可以先用 SQL 快速探索，再用 DataFrame API 固化到特征工程函数中。

from pyspark.sql import functions as F

df = spark.table("silver.user_events")

# DataFrame API：按用户聚合行为指标
user_feat = (df
    .groupBy("user_id")
    .agg(
        F.count("*").alias("event_cnt"),
        F.sum(F.when(F.col("event_type") == "click", 1).otherwise(0)).alias("click_cnt"),
        F.avg("stay_seconds").alias("avg_stay"),
        F.max("event_time").alias("last_event_time")
    ))

等价的 SQL 写法如下，结果完全一致：

SELECT user_id,
       COUNT(*) AS event_cnt,
       SUM(CASE WHEN event_type = 'click' THEN 1 ELSE 0 END) AS click_cnt,
       AVG(stay_seconds) AS avg_stay,
       MAX(event_time) AS last_event_time
FROM silver.user_events
GROUP BY user_id;

聚合与窗口函数：从明细到特征

聚合把多行压缩为一行，适合生成用户级、商品级的统计特征。窗口函数则在不减少行数的前提下，为每一行附加分组内的排名、累计或前后值，是构造时序特征和对比特征的关键工具。例如计算每个用户最近一次行为距今天数、按时间排序的行为序号、以及用户消费金额的累计值：

from pyspark.sql import Window

w_user = Window.partitionBy("user_id").orderBy(F.col("event_time"))

feat = (df
    .withColumn("rn", F.row_number().over(w_user))
    .withColumn("cum_amount", F.sum("amount").over(w_user))
    .withColumn("prev_stay", F.lag("stay_seconds", 1).over(w_user))
    .withColumn("days_since_last",
                F.datediff(F.current_date(), F.max("event_time").over(Window.partitionBy("user_id")))))

这里需要注意窗口定义的两个部分：partitionBy 决定分组边界，orderBy 决定组内顺序。若省略 orderBy，row_number 的结果将不确定；若把 orderBy 加在聚合窗口上，则 sum 会变成累计求和而非全组求和。这是窗口函数最常见的语义陷阱。

缺失值处理

真实数据中缺失值普遍存在。处理前应先量化缺失比例，再决定删除、填充还是保留为单独类别。Spark 提供 dropna、fillna 以及 Imputer 等工具。对于数值特征，常用中位数填充以降低异常值影响；对于类别特征，可填充为 "unknown" 并作为一个独立取值参与编码。

# 统计缺失比例
null_ratio = df.select([
    (F.sum(F.col(c).isNull().cast("int")) / F.count("*")).alias(c)
    for c in ["age", "income", "city"]
])
null_ratio.show()

# 数值列中位数填充，类别列填充 unknown
num_cols = ["age", "income"]
cat_cols = ["city"]

df_filled = (df
    .fillna("unknown", subset=cat_cols)
    .fillna({c: df.approxQuantile(c, [0.5], 0.01)[0] for c in num_cols}))

判断依据：若某列缺失比例超过约 60% 且业务上难以推断，通常直接删除该列；若缺失比例较低且分布随机，填充是更稳妥的选择。填充后应再次统计缺失比例，确认没有遗漏。

特征编码与缩放

类别特征需要转换为数值形式。低基数类别可用 StringIndexer 加 OneHotEncoder；高基数类别可考虑目标编码或哈希编码。数值特征若量纲差异大，应使用 StandardScaler 或 MinMaxScaler 统一尺度，避免距离类模型被大数值特征主导。

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler

idx = StringIndexer(inputCol="city", outputCol="city_idx", handleInvalid="keep")
ohe = OneHotEncoder(inputCol="city_idx", outputCol="city_vec")

assembler = VectorAssembler(
    inputCols=["age", "income", "city_vec"],
    outputCol="features_raw")

scaler = StandardScaler(inputCol="features_raw", outputCol="features",
                        withMean=True, withStd=True)

需要强调的是，编码器和缩放器的统计量（如均值、标准差、类别字典）必须只在训练集上拟合，再应用到验证集和测试集，否则会造成数据泄漏。在 Databricks 中通常把这一步封装进 Pipeline，由 Pipeline.fit(train_df) 统一完成拟合，再用 transform 处理其他数据集。

实践要点与判断依据

在 Databricks 中执行上述流程时，有几个值得注意的细节。第一，探索阶段应先用 display() 或 summary() 观察数据分布，再决定聚合粒度；第二，窗口函数的分区键应与最终特征表的粒度一致，例如用户级特征就按 user_id 分区；第三，编码与缩放必须放入 Pipeline，避免训练与推理阶段处理不一致；第四，生成的特征表建议写入 Delta Table，便于版本管理和后续复用。

综合来看，Spark 特征工程的核心是把原始明细逐步转换为模型可用的数值向量：先探索与清洗，再聚合与窗口构造统计特征，最后编码与缩放。每一步的输入、变换和输出都应可追溯，这样才能保证训练与线上推理的一致性。