请问如何将RDD拆分为两个或多个RDD？

Question

如何将RDD拆分为两个或多个RDD？我正在寻找一种将RDD分割成两个或多个RDD的方法。我最近看到的是ScalaSPark：将集合拆分成几个RDD？仍然是一个单一的RDD。如果您熟悉SAS，如下所示：data work.split1, work.split2;     set work.preSplit;     if (condition1)         output work.split1     else if (condition2)         output work.split2 run;这就产生了两个不同的数据集。必须立即坚持才能得到我想要的结果.。

Answer 1

一种方法是使用自定义分区程序根据筛选条件对数据进行分区。这可以通过扩展Partitioner并实现类似于RangePartitioner.

然后，可以使用映射分区从分区RDD构造多个RDD，而无需读取所有数据。

val filtered = partitioned.mapPartitions { iter => {

  new Iterator[Int](){
    override def hasNext: Boolean = {
      if(rangeOfPartitionsToKeep.contains(TaskContext.get().partitionId)) {
        false
      } else {
        iter.hasNext      }
    }

    override def next():Int = iter.next()
  }

请注意，筛选的RDD中的分区数将与分区RDD中的分区数相同，因此应该使用合并来减少这一点，并删除空分区。