造成内存溢出异常太多的地图键我有形式RDD[(Vector[(Int, Byte)], Vector[(Int, Byte)])]
这是一个PairRDD(key,value)
,其中关键是Vector[(Int, Byte)]
和价值是Vector[(Int, Byte)]
的RDD 'inRDD'
。火花
对于关键字段向量中的每个元素(Int, Byte)
以及值字段向量中的每个元素(Int, Byte)
我想在输出RDD中获得一个新的(键,值)对作为(Int, Int), (Byte, Byte)
。
这应该给我一个形式为RDD[((Int, Int), (Byte, Byte))]
的RDD。
例如,inRDD
内容也能像,
(Vector((3,2)),Vector((4,2))), (Vector((2,3), (3,3)),Vector((3,1))), (Vector((1,3)),Vector((2,1))), (Vector((1,2)),Vector((2,2), (1,2)))
这将成为
((3,4),(2,2)), ((2,3),(3,1)), ((3,3),(3,1)), ((1,2),(3,1)), ((1,2),(2,2)), ((1,1),(2,2))
我有下面的代码。
val outRDD = inRDD.flatMap {
case (left, right) =>
for ((ll, li) <- left; (rl, ri) <- right) yield {
(ll,rl) -> (li,ri)
}
}
它适用于向量在inRDD
中时尺寸较小的情况。但是当向量中有很多元素时,我得到了out of memory exception
。增加可用存储器 的火花只能解决较小的输入,并且对于更大的输入又会出现错误。 看起来我正试图在内存中组装一个巨大的结构。我无法以任何其他方式重写此代码。
我已经实施了与java in hadoop
类似的逻辑如下。
for (String fromValue : fromAssetVals) {
fromEntity = fromValue.split(":")[0];
fromAttr = fromValue.split(":")[1];
for (String toValue : toAssetVals) {
toEntity = toValue.split(":")[0];
toAttr = toValue.split(":")[1];
oKey = new Text(fromEntity.trim() + ":" + toEntity.trim());
oValue = new Text(fromAttr + ":" + toAttr);
outputCollector.collect(oKey, oValue);
}
}
但是当我在spark中尝试类似的东西时,我得到了嵌套的rdd异常。
我该如何有效地使用spark using scala
?
你有没有尝试解决这个增加分区的数量? – BlackBear
@BlackBear是的。但是这并没有帮助。 – CRM