Spark ML-无法使用 MatrixFactorizationModel 加载模型

Question

我正在尝试使用 Spark 协同过滤来实现推荐系统。

首先我准备模型并保存到磁盘：

MatrixFactorizationModel model = trainModel(inputDataRdd);  
model.save(jsc.sc(), "/op/tc/model/");

当我使用单独的进程加载模型时，程序失败并出现以下异常：
代码：

   static JavaSparkContext jsc ;
    private static Options options;
    static{
        SparkConf conf = new SparkConf().setAppName("TC recommender application");
        conf.set("spark.driver.allowMultipleContexts", "true");
        jsc= new JavaSparkContext(conf);
     }
MatrixFactorizationModel model = MatrixFactorizationModel.load(jsc.sc(),
                "/op/tc/model/");

例外：

线程“主”java.io.IOException 中的异常：不是文件： maprfs:/op/tc/模型/数据 在 org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:324) 在 org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1952) 在 org.apache.spark.rdd.RDD$$anonfun$aggregate$1.apply(RDD.scala:1114) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111) 在 org.apache.spark.rdd.RDD.withScope(RDD.scala:316) 在 org.apache.spark.rdd.RDD.aggregate(RDD.scala:1107) 在 org.apache.spark.mllib.recommendation.MatrixFactorizationModel.countApproxDistinctUserProduct(MatrixFactorizationModel.scala:96) 在 org.apache.spark.mllib.recommendation.MatrixFactorizationModel.predict(MatrixFactorizationModel.scala:126) 在 com.aexp.cxp.recommendation.ProductRecommendationIndividual.main(ProductRecommendationIndividual.java:62) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:497) 在 org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:742) 在 org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181) 在 org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我需要设置什么配置来加载模型吗？任何建议都会有很大帮助。

Answer 1

在 Spark 和任何其他分布式计算框架中一样，在尝试调试代码时了解代码运行的位置很重要。访问各种类型也很重要。例如，在 YARN 中，您将拥有：

• 如果您自己记录主日志
• 汇总的从属日志（感谢 YARN，有用的功能！）
• YARN 节点管理器（例如会告诉您为什么容器被杀死等）
• 等

如果您从一开始就没有找到正确的位置，那么深入研究 Spark 问题可能会非常耗时。现在更具体地说，关于这个问题，您有一个清晰的堆栈跟踪，但并非总是如此，因此您应该利用它来发挥自己的优势。

堆栈跟踪的顶部是

线程“主”java.io.IOException 中的异常：不是文件： maprfs:/op/tc/model/data 在 org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:324) 在 org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 在

如您所见，Spark 作业在失败时正在执行map 操作。谁执行 map ？从站，因此您必须确保您的文件在所有从站上都可用，而不仅仅是在主站上。

更一般地说，您总是需要在头脑中清楚地区分您为 master 编写的代码和您为 slave 编写的代码。这将帮助您检测此类交互，以及对不可序列化对象的引用和此类常见错误。