PySpark3.2.0下ASL0.9.0使用ParamGridBuilder及ALS初始化报错

问题产生原因 #

你遇到的三个异常本质上是导入的ALS实现版本不匹配导致的：

• 你当前导入的是pyspark.mllib.recommendation包下基于RDD API的旧版ALS工具类，这个类本身设计为无参构造，所有训练参数仅支持在调用train()方法时作为入参传入，类实例不会绑定rank、iterations这类参数属性，因此实例化传参时报TypeError、访问实例参数时报AttributeError都是符合该类设计逻辑的正常表现。
• ParamGridBuilder是Spark ML Pipeline体系的专属工具，仅能配合实现了pyspark.ml.param.Params接口的估计器/转换器使用，要求传入的参数是类上定义的Param类型属性，旧版RDD API的ALS完全没有适配这套接口，无法直接搭配使用。
• 额外注意：旧版RDD API的ALS正则化参数名为lambda_，ML体系新版ALS的正则化参数名为regParam，迭代次数参数旧版叫iterations、新版叫maxIter，参数名不通用。

可行实现方案 #

方案1：切换为ML体系的ALS实现（推荐） #

替换导入路径为基于DataFrame API的新版ALS，原生支持构造传参、ParamGridBuilder网格搜索、交叉验证等Pipeline能力，适配你预期的使用逻辑：

# 注意导入路径为pyspark.ml而非pyspark.mllib
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import RegressionEvaluator

# 实例化时可直接传入默认配置
als = ALS(
    userCol="userId",
    itemCol="movieId",
    ratingCol="rating",
    coldStartStrategy="drop" # 避免冷启动用户/物品预测值为NaN影响评估
)

# 可正常构建参数网格，注意参数名和旧版RDD API的区别
param_grid = ParamGridBuilder() \
    .addGrid(als.rank, [10, 50, 100, 150]) \
    .addGrid(als.maxIter, [20]) \
    .addGrid(als.regParam, [.01, .05, .1, .15]) \
    .build()

# 后续可直接搭配交叉验证完成模型选优
evaluator = RegressionEvaluator(
    metricName="rmse",
    labelCol="rating",
    predictionCol="prediction"
)
cv = CrossValidator(
    estimator=als,
    estimatorParamMaps=param_grid,
    evaluator=evaluator,
    numFolds=3
)
# 输入要求为DataFrame格式，而非旧版的RDD[Rating]格式
cv_model = cv.fit(train_df)

方案2：保留旧版RDD API，手动实现参数遍历 #

如果因历史代码兼容要求必须使用mllib下的RDD版ALS，无法直接使用ParamGridBuilder，可自行遍历参数组合完成网格搜索：

from pyspark.mllib.recommendation import ALS
from itertools import product
import math

# 定义待搜索的参数空间
rank_list = [10, 50, 100, 150]
iter_list = [20]
reg_param_list = [.01, .05, .1, .15]

best_rmse = float("inf")
best_model = None
best_params = None

# 遍历所有参数组合训练、评估
for rank, iterations, lambda_ in product(rank_list, iter_list, reg_param_list):
    model = ALS.train(
        train_rdd,
        rank=rank,
        iterations=iterations,
        lambda_=lambda_,
        blocks=-1,
        nonnegative=False,
        seed=42
    )
    # 计算验证集RMSE
    predictions = model.predictAll(
        test_rdd.map(lambda x: (x.user, x.product))
    ).map(lambda r: ((r.user, r.product), r.rating))
    rates_and_preds = test_rdd.map(
        lambda r: ((r.user, r.product), r.rating)
    ).join(predictions)
    rmse = math.sqrt(rates_and_preds.map(lambda x: (x[1][0] - x[1][1])**2).mean())
    
    # 保留最优模型
    if rmse < best_rmse:
        best_rmse = rmse
        best_model = model
        best_params = {"rank": rank, "iterations": iterations, "lambda_": lambda_}

内容的提问来源于stack exchange，提问作者Katereena Tran