4. 特征工程核心流程与Spark DataFrame操作

在Databricks上做特征工程，核心工作几乎都围绕Spark DataFrame展开。可以把特征工程理解为一条“原始表 → 特征表”的转换链：从原始数据中选择需要的列，过滤掉无效或异常样本，派生新的特征列，再通过聚合和窗口函数把跨行信息压缩成单行特征，最后用UDF或Pandas UDF处理Spark SQL难以表达的复杂逻辑。整条链路的产物应当是可复用、可测试、可版本化的特征转换逻辑，而不是一次性写在Notebook里的临时代码。

选择与过滤：先把数据范围收敛

选择（select）决定进入特征表的列，过滤（filter/where）决定进入特征表的行。两者共同定义了后续所有计算的数据边界。实践中建议先做列裁剪再做行过滤，因为列裁剪能减少后续shuffle的数据量。例如从订单表构造用户购买力特征时，先选出user_id, order_id, amount, ts, status，再过滤status = 'PAID'且amount > 0，避免把退款单和异常金额带入统计。

from pyspark.sql import functions as F

base = (spark.table("prod.silver.orders")
        .select("user_id", "order_id", "amount", "ts", "status")
        .filter((F.col("status") == "PAID") & (F.col("amount") > 0)))

过滤条件要尽量下推到数据源，Databricks的Delta表支持分区裁剪和文件跳过，把时间范围条件写在filter里能显著减少扫描量。需要注意filter与where完全等价，但多个条件用&连接时每个条件都要加括号，否则会因运算符优先级产生错误结果。

派生列：把业务语义写进特征

派生列（withColumn）是特征工程中出现频率最高的操作。它把原始字段转换成模型可用的数值或类别特征。常见派生包括：金额分桶、时间特征提取、比率计算、字符串规范化。例如从ts派生小时、星期几，从amount派生对数金额：

feat = (base
        .withColumn("hour", F.hour("ts"))
        .withColumn("dow", F.dayofweek("ts"))
        .withColumn("log_amount", F.log1p("amount"))
        .withColumn("amount_bucket",
                    F.when(F.col("amount") < 50, "low")
                     .when(F.col("amount") < 500, "mid")
                     .otherwise("high")))

派生列应遵循“一列一语义”的原则，避免把多个含义塞进同一列。同时要注意withColumn会重建整个逻辑计划，连续几十次调用会拖慢解析，建议合并为一次select表达式，或使用withColumns批量派生。

聚合与窗口函数：把跨行信息压成单行

聚合（groupBy + agg）把多行压缩为一行，适合构造用户级、商品级的统计特征，如近30天订单数、平均客单价、最大单笔金额。窗口函数（Window）则在不减少行数的前提下，为每行附加基于分组的统计值，适合构造“当前订单相对该用户历史”的对比特征。

from pyspark.sql import Window

user_agg = (feat.groupBy("user_id").agg(
    F.count("order_id").alias("order_cnt_30d"),
    F.avg("amount").alias("avg_amount"),
    F.max("amount").alias("max_amount")))

w = Window.partitionBy("user_id").orderBy("ts").rowsBetween(-5, 0)
feat_win = feat.withColumn("amount_ma6", F.avg("amount").over(w))

窗口定义中的partitionBy决定分组键，orderBy决定行序，rowsBetween决定窗口范围。这里rowsBetween(-5, 0)表示当前行及前5行，用于计算滑动均值。窗口函数会触发shuffle，分区键的选择直接影响性能，通常选择基数适中且分布均匀的列作为partitionBy键。

UDF与Pandas UDF：处理Spark SQL表达不了的逻辑

当逻辑无法用内置函数表达时，才考虑UDF。普通UDF逐行执行，Python与JVM之间需要序列化，性能较差。Pandas UDF（向量化UDF）以Arrow为传输格式，按批次处理，性能通常比普通UDF高一个数量级，是Databricks上推荐的方式。

from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf("double")
def geo_mean(s: pd.Series) -> pd.Series:
    return s.apply(lambda x: x ** 0.5)

feat2 = feat.withColumn("sqrt_amount", geo_mean(F.col("amount")))

使用Pandas UDF时要注意：输入输出类型必须显式声明；函数内不要做跨行聚合，跨行逻辑应交给窗口或聚合；批次大小可通过spark.sql.execution.arrow.maxRecordsPerBatch调整。UDF的另一个代价是难以被Catalyst优化器下推，因此能用内置函数就不要用UDF。

构建可复用的特征转换逻辑

把上述操作封装成函数或类，是特征工程从“一次性脚本”走向“可复用资产”的关键。推荐做法是：每个特征组一个函数，输入DataFrame、输出DataFrame，参数通过配置传入，函数内部不读取外部表。这样既方便单元测试，也方便在训练和推理时复用同一套逻辑，避免训练/服务偏差。

def build_user_features(df, lookback_days=30):
    w = Window.partitionBy("user_id").orderBy("ts").rowsBetween(-lookback_days, 0)
    return (df
            .withColumn("log_amount", F.log1p("amount"))
            .withColumn("amount_ma", F.avg("amount").over(w))
            .groupBy("user_id")
            .agg(F.count("order_id").alias("cnt"),
                 F.avg("log_amount").alias("avg_log_amount")))

在Databricks中，这类函数可以放入Feature Store的转换逻辑，或注册为Unity Catalog中的函数，供不同Notebook和Job调用。配合spark.conf.set("spark.sql.shuffle.partitions", ...)等参数调优，可以在保证逻辑一致的前提下控制执行成本。

总结来看，Spark DataFrame的特征工程是一条从选择、过滤、派生、聚合、窗口到UDF的转换链。每一步都对应明确的输入输出和性能代价。把这条链封装成可复用函数，配合Databricks的Feature Store和Unity Catalog，才能让特征逻辑在训练、批推理和在线服务之间保持一致，真正支撑起可维护的模型训练流水线。