JavaRDD 上每个操作的 Apache Spark 计时

Question

问题：这是测试构建 RDD 所用时间的有效方法吗？

我在这里做两件事。基本方法是我们有 M 个我们称为 DropEvaluation 的实例和 N 个 DropResults。我们需要将每个 N DropResult 与每个 M DropEvaluations 进行比较。每个 N 必须被每个 M 看到，最终给我们 M 个结果。

如果我在构建 RDD 后不使用 .count()，驱动程序会继续执行下一行代码，并表示几乎没有时间构建需要 30 分钟才能构建的 RDD。

我只是想确保我没有遗漏一些东西，比如 .count() 可能需要很长时间？我猜要计时 .count() 我必须修改 Spark 的源代码？

M = 1000 或 2000。N = 10^7。

这实际上是一个笛卡尔问题——选择累加器是因为我们需要在适当的位置写入每个 M。构建完整的笛卡尔 RDD 也会很丑。

我们构建了一个 M 累加器列表（不能在 Java 中做一个列表累加器，对吧？）。然后我们用 foreach 循环遍历 RDD 中的 N 中的每一个。

澄清问题：总时间测量正确，我在问 RDD 上的 .count() 是否强制 Spark 等到 RDD 完成才能运行计数。 .count() 时间是否重要？

这是我们的代码：

// assume standin exists and does it's thing correctly

// this controls the final size of RDD, as we are not parallelizing something with an existing length
List<Integer> rangeN = IntStream.rangeClosed(simsLeft - blockSize + 1, simsLeft).boxed().collect(Collectors.toList());

// setup bogus array of size N for parallelize dataSetN to lead to dropResultsN       
JavaRDD<Integer> dataSetN = context.parallelize(rangeN);

// setup timing to create N
long NCreationStartTime = System.nanoTime();

// this maps each integer element of RDD dataSetN to a "geneDropped" chromosome simulation, we need N of these:
JavaRDD<TholdDropResult> dropResultsN = dataSetN.map(s -> standin.call(s)).persist(StorageLevel.MEMORY_ONLY());

// **** this line makes the driver wait until the RDD is done, right?
long dummyLength = dropResultsN.count();


long NCreationNanoSeconds = System.nanoTime() - NCreationStartTime;
double NCreationSeconds = (double)NCreationNanoSeconds / 1000000000.0;
double NCreationMinutes = NCreationSeconds / 60.0;

logger.error("{} test sims remaining", simsLeft);

// now get the time for just the dropComparison (part of accumulable's add)
long startDropCompareTime = System.nanoTime();

// here we iterate through each accumulator in the list and compare all N elements of dropResultsN RDD to each M in turn, our .add() is a custom AccumulableParam
for (Accumulable<TholdDropTuple, TholdDropResult> dropEvalAccum : accumList) {
    dropResultsN.foreach(new VoidFunction<TholdDropResult>() {
                    @Override
                    public void call(TholdDropResult dropResultFromN) throws Exception {
                            dropEvalAccum.add(dropResultFromN);
                    }
                });
            }

    // all the dropComparisons for all N to all M for this blocksize are done, check the time...
   long dropCompareNanoSeconds = System.nanoTime() - startDropCompareTime;
   double dropCompareSeconds = (double)dropCompareNanoSeconds / 1000000000.0;
    double dropCompareMinutes = dropCompareSeconds / 60.0;

    // write lines to indicate timing section
    // log and write to file the time for the N-creation

    ...

} // end for that goes through dropAccumList

Answer 1

Spark 程序是惰性的，除非您在 RDD 上调用所有类似 count 的操作，否则它不会运行。您可以在Spark's document中找到常用操作列表

// **** this line makes the driver wait until the RDD is done, right?
long dummyLength = dropResultsN.count();

是的，在这种情况下，count 强制计算 dropResultsN，因此需要很长时间。如果你再做一次count，它会很快返回，因为 RDD 已经被计算和缓存了。