无法更改 RDD 的存储级别

Question

以下 Spark 代码：

val model = ALS.trainImplicit(ratings = ratingsRDD,
                              rank = rank,
                              iterations = numIterations,
                              lambda = lambda,
                              alpha = alpha)  

model.productFeatures.cache()

val modelSubsetRDD = new MatrixFactorizationModel(
  rank = rank,
  userFeatures = model.productFeatures,
  productFeatures = model.productFeatures)

引发以下异常：

在 RDD 已经分配了一个之后，不能更改它的存储级别 等级

StorageLevel.MEMORY_ONLY 引发了相同的异常。

另一方面，以下代码可以正常工作：

    val model = ALS.trainImplicit(ratings = ratingsRDD,
                              rank = rank,
                              iterations = numIterations,
                              lambda = lambda,
                              alpha = alpha)  
    val modelSubsetRDD = new MatrixFactorizationModel(
      rank = rank,
      userFeatures = model.userFeatures,
      productFeatures = model.productFeatures)

    model.userFeatures.persist(StorageLevel.MEMORY_ONLY)
    model.productFeatures.persist(StorageLevel.MEMORY_ONLY)

注意到这次userFeatures 和productFeatures 被设置为模型的两个不同成员。但是，我不确定为什么会这样。

Answer 1

您可能从代码中的其他地方获得了一些持久性？在返回模型之前不确定ALS.trainImplicit 在做什么。

调用cache() 会将RDD 存储在MEMORY_ONLY 中，而调用persist 允许您更改缓存类型。所以我猜这个 RDD 已经在其他地方持久化了，你正试图用cache() 重新持久化它，这就是问题所在。但是，使用persist 更改持久性类型是完全可以接受的。

编辑：

试试下面的代码：

val model = ALS.trainImplicit(ratings = ratingsRDD,
                              rank = rank,
                              iterations = numIterations,
                              lambda = lambda,
                              alpha = alpha)  
if(model.productFeatures.getStorageLevel() == StorageLevel.NONE)
    model.productFeatures.cache()

val modelSubsetRDD = new MatrixFactorizationModel(
  rank = rank,
  userFeatures = model.productFeatures,
  productFeatures = model.productFeatures)

这应该避免您尝试缓存已经缓存的内容（在内存或磁盘中）。