scala - 太多 map 键导致 Spark 内存不足异常

Question

我有一个 RDD 'inRDD'形式 RDD[(Vector[(Int, Byte)], Vector[(Int, Byte)])]这是一个 PairRDD(key,value)其中关键是 Vector[(Int, Byte)]值是 Vector[(Int, Byte)] .

对于每个元素 (Int, Byte)在关键字段的向量中，每个元素(Int, Byte)在值字段的向量中，我想在输出 RDD 中获得一个新的 (key,value) 对为 (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 .增加可用内存
spark 只能解决较小的输入，而对于更大的输入，错误再次出现。
看起来我正试图在内存中组装一个巨大的结构。我无法以任何其他方式重写此代码。

我已经用 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 有效地做到这一点?

Answer 1

好吧，如果笛卡尔积是唯一的选择，你至少可以让它更懒一点:

inRDD.flatMap { case (xs, ys) =>
  xs.toIterator.flatMap(x => ys.toIterator.map(y => (x, y)))
}

您也可以在 Spark 级别处理此问题

import org.apache.spark.RangePartitioner

val indexed = inRDD.zipWithUniqueId.map(_.swap)
val partitioner = new RangePartitioner(indexed.partitions.size, indexed)
val partitioned = indexed.partitionBy(partitioner)

val lefts = partitioned.flatMapValues(_._1)
val rights = partitioned.flatMapValues(_._2)

lefts.join(rights).values