apache spark - 迭代地跳过并从 RDD 中获取

Question

给定一个 RDD，对它进行排序然后在离散大小的块中使用它的最佳方法是什么？例如：

  JavaRDD<Integer> baseRdd = sc.parallelize(Arrays.asList(1,2,5,3,4));

  JavaRDD<Integer> sorted = baseRdd.sortBy(x -> x, true, 5);   

  // returns 1, 2   
  List<Integer> first = sorted.take(2);

  // returns 1, 2.  How to skip 2 and then take?
  List<Integer> second = sorted.take(2);

我真正想要的是在第一次调用take(2) 时使用1, 2，然后将某种“跳过”参数传递给第二个take(2) 以返回3, 4？

由于当前 RDD 功能中似乎不存在“跳​​过”功能，将已排序的 RDD 拆分为可以独立操作的已知大小的块的最有效方法是什么？

Answer 1

为了提高效率，不要忘记您可以随时缓存您的 RDD。这将避免每次调用take 时都从文本文件中重新计算已排序的 RDD。由于我们将多次使用sorted RDD，我们将其缓存：

JavaRDD<Integer> sorted = baseRdd.sortBy(x -> x, true, 5).cache();

---

然后要从给定索引中获取元素到另一个索引，我们可以结合zipWithIndex 和filter。 zipWithIndex 将 RDD 转换为元组的 RDD，其中元组的第一部分是已排序 RDD 的元素，第二部分是它的索引。一旦我们有了这些索引记录，我们就可以根据它们的索引过滤它们（假设 offset = 2 和 window = 2）：

List<Integer> nth =
  sorted.zipWithIndex()
  .filter(x -> x._2() >= offset && x._2() < offset + window)
  .map(x -> x._1())
  .collect();

返回：

[3, 4]

---

最终的结果是：

JavaRDD<Integer> sorted = baseRdd.sortBy(x -> x, true, 5).zipWithIndex().cache();

Integer offset = 2;
Integer window = 2;

List<Integer> nth =
  sorted
  .filter(x -> x._2() >= offset && x._2() < offset + window)
  .map(x -> x._1())
  .collect();

这里我只在使用 index 压缩后才缓存 rdd，以免每次在不同窗口上执行此操作时都执行压缩部分。

然后，您可以根据您要如何创建不同的窗口列表，将此 nth creation sn-p 映射到循环或映射中。

Answer 2

rdd1=sc.parallelize((1,2,3,4,5,6,7,8))
rdd2=rdd1.take(2)

现在你根据 rdd2 过滤你的初始 rdd

rdd1.filter(lambda line:line not in rdd2).take(2)

这给出 [3, 4]

使用 PySpark