SparkML CrossValidator调用setParallelism报错求助

嘿，我来帮你搞定这个问题！

你遇到的value setParallelism is not a member of org.apache.spark.ml.tuning.CrossValidator错误，核心原因很明确：你当前使用的Spark版本里，CrossValidator类根本没有setParallelism这个方法。

这个方法是Spark 3.0.0版本之后才新增的功能，用来设置交叉验证时并行训练的任务数。如果你的Spark版本还停留在2.x系列，自然会触发这个报错——哪怕你是跟着官方教程操作，也要注意教程对应的Spark版本是不是和你本地/集群的版本匹配哦！

给你两个解决方案，按需选择：

方案一：升级Spark到3.0及以上版本 #

如果条件允许（比如本地测试、集群支持升级），升级后就可以直接调用setParallelism了。记得要确保正确导入CrossValidator类：

import org.apache.spark.ml.tuning.CrossValidator

调用方式也很简单，比如设置并行数为2：

val cv = new CrossValidator()
  // 其他必要设置：setEstimator、setEvaluator、setNumFolds等
  .setParallelism(2)

方案二：不升级版本，间接控制并行度 #

如果没法升级Spark，你可以通过调整Spark的全局配置参数来间接控制交叉验证的并行能力：

1. 设置默认并行度：

val spark = SparkSession.builder()
  .appName("YourAppName")
  .config("spark.default.parallelism", "4") // 根据你的机器/集群资源调整数值
  .getOrCreate()

2. 调整shuffle分区数（交叉验证过程中会涉及shuffle操作）：

spark.conf.set("spark.sql.shuffle.partitions", "4")

最后给你补一段完整的、适配不同Spark版本的示例代码，方便你对照：

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
import org.apache.spark.sql.SparkSession

object CrossValidatorDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("CrossValidatorTest")
      .master("local[*]") // 本地测试用，集群环境请删除这一行
      .getOrCreate()

    // 模拟训练数据
    val trainingData = spark.createDataFrame(Seq(
      (0L, "a b c d e spark", 1.0),
      (1L, "b d", 0.0),
      (2L, "spark f g h", 1.0),
      (3L, "hadoop mapreduce", 0.0),
      (4L, "b spark who", 1.0),
      (5L, "g d a y", 0.0),
      (6L, "spark fly", 1.0),
      (7L, "was mapreduce", 0.0),
      (8L, "e spark program", 1.0),
      (9L, "a e c l", 0.0)
    )).toDF("id", "text", "label")

    // 构建ML Pipeline
    val tokenizer = new Tokenizer().setInputCol("text").setOutputCol("words")
    val hashingTF = new HashingTF().setInputCol(tokenizer.getOutputCol).setOutputCol("features")
    val lr = new LogisticRegression().setMaxIter(10)
    val pipeline = new Pipeline().setStages(Array(tokenizer, hashingTF, lr))

    // 构建参数网格
    val paramGrid = new ParamGridBuilder()
      .addGrid(hashingTF.numFeatures, Array(10, 100, 1000))
      .addGrid(lr.regParam, Array(0.1, 0.01))
      .build()

    // 初始化交叉验证器
    val cv = new CrossValidator()
      .setEstimator(pipeline)
      .setEvaluator(new BinaryClassificationEvaluator())
      .setEstimatorParamMaps(paramGrid)
      .setNumFolds(3) // 3折交叉验证
      // 仅Spark 3.0+版本可用：设置并行训练任务数
      // .setParallelism(2)

    // 训练模型
    val cvModel = cv.fit(trainingData)

    // 后续的模型评估、预测等操作...

    spark.stop()
  }
}

内容的提问来源于stack exchange，提问作者Leo R