如何替换 RDD 的元素

Question

我有一个这样的 RDD：

JavaPairRDD<SubspaceFlag, Iterable<Point>> flagPointPairs

示例值：

(01), [(5, 5), (6, 1), (7, 2), (9, 4)]

Point是二维空间中的一个(x, y)点

我需要从Tuple2 的Iterable<Point> 部分中删除一些元素。例如，我只需要保留支配其他(x1 <= x2 AND y1 < y2 OR y1 <= y2 AND x1 < x2)的点。

最后我留下了一个临时列表[(5, 5), (6, 1)]，我想将其转换回JavaPairRDD<SubspaceFlag, Iterable<Point>> -> (01), [(5, 5), (6, 1)]。

我知道如何访问Iterable<Point>，我知道如何从Iterable<Point> 中创建我感兴趣的较小列表，但我不知道如何将该列表返回到伴随@987654332 的RDD @。

Answer 1

RDD 是不可变结构，因此您不会替换 RDD 中的元素，而是将 RDD 转换为另一个 RDD。 在这种特定情况下，rdd.map 转换将允许您对 RDD 的每个元素应用一个函数，从而生成一个新的 RDD。

给定

val flagPointPairsRdd:RDD[SubspaceFlag, Iterable[Point]] = ???

还有一个功能：

def dominants(points: Iterable[Point]):Iterable[Point] = ??? //Your impl here

那么你可以通过以下方式获得想要的结果：

val dominatingPairsRdd = flagPointPairs.map{case (flag, points) => (flag, dominants(points))}

等效的 Java 代码非常相似（只是更冗长），并留给读者作为练习。

Answer 2

@maasg 的答案几乎是我需要知道的（为了返回相同类型的 RDD，我应该只映射它）但由于我的问题是在 Java 中，而在 Java 中的答案完全不同，我将发布Java 中的答案也是如此。

有这样的RDD：

JavaPairRDD<SubspaceFlag, Iterable<Point>> flagPointPairs = ...

为了返回相同类型但具有不同值的 RDD，首先您需要这样的方法：

private Iterable<Point> localSkylines(Iterable<Point> pointIterable) {
        List<Point> localSkylines = new ArrayList<>();

        // find local skylines here

        return localSkylines;
    }

然后你可以像这样制作新的RDD：

JavaPairRDD<PointFlag, Iterable<Point>> dominatingPairs = flagPointPairs.mapToPair(fp -> new Tuple2(fp._1, localSkylines(fp._2)));

这里的主要区别有两个：

1. 在 Java 中，为了取回 JavaPairRDD，您必须使用 .mapToPair() 并使用 new Tuple2(left, right)
2. 要返回 Iterable<Point>，您无需转换 List<Point>，因为它已经是 Iterable<Point>。