Spark MLlib推荐任务中DataFrame操作触发Task Not Serializable异常

问题场景 #

你在使用Spark MLlib的ALS推荐模块时，调用recommendForAllUsers生成推荐结果后，执行show()、collect()等操作时触发了Task Not Serializable异常，核心报错指向scala.reflect.api.TypeTags$PredefTypeCreator不可序列化。你的实现代码大致如下：

object RecommendBasedView extends App { new MachineLearning().doIt } 
class MachineLearning extends Serializable { 
  def doIt() = { 
    val spark = SparkSession.builder.master("local").appName("RecommendBasedView").getOrCreate() 
    val data = spark.read.option("header","true").option("inferSchema","true").format("csv").load("productionView.csv") 
    val Array(training, test) = data.randomSplit(Array(0.8, 0.2)) 
    val als = new ALS() 
      .setMaxIter(5) 
      .setRegParam(0.01) 
      .setUserCol("userId") 
      .setItemCol("productionId") 
      .setRatingCol("count") 
    val model = als.fit(training) 
    model.setColdStartStrategy("drop") 
    val predictions = model.transform(test) 
    val evaluator = new RegressionEvaluator() 
      .setMetricName("rmse") 
      .setLabelCol("count") 
      .setPredictionCol("prediction") 
    val rmse = evaluator.evaluate(predictions) 
    println(s"Root-mean-square error = $rmse") 
    val userRecs = model.recommendForAllUsers(10) 
    userRecs.show() 
  } 
}

异常原因分析 #

这个异常的根源在于：你把整个机器学习逻辑封装在了实现Serializable的MachineLearning类中，而Spark在执行recommendForAllUsers时，其内部的TopByKeyAggregator组件依赖了Scala的TypeTag，这个TypeTag的底层实现类PredefTypeCreator并不支持序列化。当Spark尝试序列化MachineLearning实例的上下文时，就会把这个不可序列化的对象带进去，从而触发报错。

另外，部分旧版本的Spark（比如2.x早期版本）本身存在MLlib推荐模块的序列化bug，也会导致这个问题。

解决方案 #

这里提供两个有效的解决办法，你可以根据自己的情况选择：

方案1：重构代码结构，移除不必要的Serializable类

Driver端的代码不需要实现Serializable接口，你可以把所有逻辑直接放到RecommendBasedView对象中，避免Spark尝试序列化整个类实例：

object RecommendBasedView extends App {
  // 直接在App对象中执行所有逻辑
  val spark = SparkSession.builder.master("local").appName("RecommendBasedView").getOrCreate()
  val data = spark.read.option("header","true").option("inferSchema","true").format("csv").load("productionView.csv")
  val Array(training, test) = data.randomSplit(Array(0.8, 0.2))
  
  val als = new ALS()
    .setMaxIter(5)
    .setRegParam(0.01)
    .setUserCol("userId")
    .setItemCol("productionId")
    .setRatingCol("count")
    
  val model = als.fit(training)
  model.setColdStartStrategy("drop")
  
  val predictions = model.transform(test)
  val evaluator = new RegressionEvaluator()
    .setMetricName("rmse")
    .setLabelCol("count")
    .setPredictionCol("prediction")
    
  val rmse = evaluator.evaluate(predictions)
  println(s"Root-mean-square error = $rmse")
  
  val userRecs = model.recommendForAllUsers(10)
  userRecs.show()
}

方案2：升级Spark版本到稳定版

如果你的Spark版本比较旧（比如2.2.x及更早），可以考虑升级到2.4.x的稳定版本或者3.x系列，这些版本修复了MLlib推荐模块中TypeTag序列化的bug，能直接解决这个问题。

验证步骤 #

修改代码或升级Spark后，重新运行程序：

• 观察是否还会抛出Task Not Serializable异常
• 确认userRecs.show()能正常打印出推荐结果

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