PySpark DataFrame窗口分区后执行标准化操作报错排查

问题 #

针对每个article_id，若其首个transaction_date晚于2022年1月1日，则新增new_item列赋值为1，否则为0。通过窗口函数实现该逻辑后，后续执行标准化步骤时抛出异常，无窗口分区代码时可正常运行。

相关代码 #

窗口分区新增列代码

windowSpec  = Window.partitionBy("article_id").orderBy("transaction_date")

feature_grid = feature_grid.withColumn("row_number",row_number().over(windowSpec)) \
.withColumn('new_item', 
            when(
              (f.col('row_number') == 1) & (f.col('transaction_date') >= ‘01.01.2022’), 1) .otherwise(0))\
.drop('row_number')

聚类前特征处理代码

from pyspark.ml.feature import VectorAssembler
input_cols = feature_grid.columns

assemble=VectorAssembler(inputCols= input_cols, outputCol='features')
assembled_data=assemble.transform(feature_grid)

from pyspark.ml.feature import StandardScaler
scale=StandardScaler(inputCol='features',outputCol='standardized')
data_scale=scale.fit(assembled_data)
data_scale_output=data_scale.transform(assembled_data)
display(data_scale_output)

报错信息 #

org.apache.spark.SparkException: Job aborted due to stage failure:
Task 0 in stage 182.0 failed 4 times, most recent failure: Lost task
0.3 in stage 182.0 (TID 3635) (10.205.234.124 executor 1): org.apache.spark.SparkException: Failed to execute user defined
function (VectorAssembler$$Lambda$3621/907379691

错误原因与修正方案 #

核心错误原因 #

1. 日期比较逻辑错误：代码中直接用字符串‘01.01.2022’和transaction_date列比较，若transaction_date是日期类型，会触发隐式类型转换异常；若为字符串类型，格式不匹配会导致比较结果不符合预期，最终生成的new_item列可能存在脏数据，干扰后续特征组装。
2. VectorAssembler输入列包含非数值类型：input_cols = feature_grid.columns将所有列（包括article_id、transaction_date这类字符串/日期列）都作为特征输入，但VectorAssembler仅支持数值型、布尔型、向量型列，非数值列会导致组装失败。

修正步骤 #

1. 修复日期比较逻辑：将日期字符串转换为Spark标准日期类型后再做比较：

from pyspark.sql import functions as f
from pyspark.sql.window import Window

windowSpec = Window.partitionBy("article_id").orderBy("transaction_date")

feature_grid = feature_grid.withColumn("row_number", f.row_number().over(windowSpec)) \
    .withColumn('new_item', 
                f.when(
                  (f.col('row_number') == 1) & (f.col('transaction_date') >= f.to_date(f.lit('2022-01-01'))), 1)
                .otherwise(0)) \
    .drop('row_number')

2. 筛选VectorAssembler的合法输入列：排除非数值类型的列，只保留可作为特征的数值/布尔列：

from pyspark.ml.feature import VectorAssembler

# 排除非特征列，根据实际数据调整列名
input_cols = [col for col in feature_grid.columns if col not in ['article_id', 'transaction_date']]

assemble = VectorAssembler(inputCols=input_cols, outputCol='features')
assembled_data = assemble.transform(feature_grid)

3. 提前验证数据类型：执行特征组装前，可先查看表结构确认输入列类型：

feature_grid.printSchema()

内容的提问来源于stack exchange，提问作者snigdha mohapatra