val itemFactors = rddItems.mapValues(newFactors =>
Vector(Array.fill(2){math.random})
)
然后,我将该RDD与其他RDD连接并缓存:
val finalRDD = itemFactors.join(rddItemsUsers).map{
case(itemid, (itemVector, ((userid, rating), userVector))) =>
(itemid, itemVector, userid, userVector, rating)}.cache
然后我对finalRDD
中保存的数据进行计算:
sqrt(finalRDD.aggregate(0.0)((accum, item) =>
accum + pow(item._5 - item._4.dot(item._2), 2), _ + _) / finalRDD.count)
我从控制台多次调用代码的最后一部分,即
sqrt(...)
,但每次都得到不同的结果 - 这是不希望出现的,因为我什么也没有改变!有两种方法可以解决这个问题(即使我得到一致的结果):
不要使用
math.random
初始化itemFactors
,而是使用一个固定数值(例如1.0)填充数组。使用
itemFactors.cache
。
itemFactors
时都会调用math.random
并创建一个新的数字 - 这将影响我的计算。因此,使用固定数字填充数组会产生一致的结果。但是,大问题和我不理解的部分是:我正在缓存
finalRDD
,这是计算执行的内容,由itemFactors
组成,所以节点只被访问一次,那么itemFactor
的数组填充什么内容似乎并不重要?我以为我开始理解继承关系了,然而这却让我感到困惑。
cache()
不能保证它总是被缓存--例如,您是否有足够的内存来保存它?它的缓存百分比是多少?如果丢失一个块,它也会被重新计算。 - Sean OwengetStorageLevel
并得到了:StorageLevel(false, true, false, true, 1)
。如果我执行getStorageLevel.description
,我会得到:Memory Deserialized 1x Replicated
。getStorageLevel.useMemory
返回True
。 - monsterrddItems
和rddItemsUsers
里面是什么。可能是小矮人。 - Daniel Darabos