Improper unpersist strategy in ml.recommendation.ASL.train

    val userIdAndFactors = userInBlocks
      .mapValues(_.srcIds)
      .join(userFactors)
      .mapPartitions({ items =>
        items.flatMap { case (_, (ids, factors)) =>
          ids.view.zip(factors)
        }
      // Preserve the partitioning because IDs are consistent with the partitioners in userInBlocks
      // and userFactors.
      }, preservesPartitioning = true)
      .setName("userFactors")
      .persist(finalRDDStorageLevel) // Missing unpersist, but hard to fix
    val itemIdAndFactors = itemInBlocks
      .mapValues(_.srcIds)
      .join(itemFactors)
      .mapPartitions({ items =>
        items.flatMap { case (_, (ids, factors)) =>
          ids.view.zip(factors)
        }
      }, preservesPartitioning = true)
      .setName("itemFactors")
      .persist(finalRDDStorageLevel) // Missing unpersist, but hard to fix
    if (finalRDDStorageLevel != StorageLevel.NONE) {
      userIdAndFactors.count()
      itemFactors.unpersist() // Premature unpersist
      itemIdAndFactors.count()
      userInBlocks.unpersist() // Lagging unpersist
      userOutBlocks.unpersist() // Lagging unpersist
      itemInBlocks.unpersist() 
      itemOutBlocks.unpersist() // Lagging unpersist
      blockRatings.unpersist() // Lagging unpersist
    }
    (userIdAndFactors, itemIdAndFactors)
  }