apache-spark - 在 Spark mapPartitions 中使用 Java 8 并行流

Question

我试图了解 Spark 并行性中 Java 8 并行流的行为。当我运行以下代码时，我期望输出大小为 listOfThings与输入大小相同。但事实并非如此，我有时会在输出中丢失项目。这种行为是不一致的。如果我只是遍历迭代器而不是使用 parallelStream ， 一切安好。每次都计算匹配。

// listRDD.count = 10
JavaRDD test = listRDD.mapPartitions(iterator -> {
    List listOfThings = IteratorUtils.toList(iterator);
    return listOfThings.parallelStream.map(
        //some stuff here
    ).collect(Collectors.toList());
});
// test.count = 9
// test.count = 10
// test.count = 8
// test.count = 7

Answer 1

这是一个很好的问题。

这是怎么回事Race Condition .当您并行化流然后流将完整列表拆分为几个相等的部分 [基于可用线程和列表大小] 然后它尝试在每个可用线程上独立运行子部分以执行工作。

但是您也在使用 apache spark，它以更快地计算工作而闻名，即通用计算引擎。 Spark 使用相同的方法 [并行化工作] 来执行操作。

现在在这个 Scenerio 中发生的事情是 Spark 已经并行化了整个工作然后在这个内部你再次并行化工作，因为竞争条件开始即 Spark 执行器开始处理工作然后你并行化工作然后流处理获得其他线程并开始处理，如果正在处理流工作的线程在 Spark 执行器完成其工作之前完成工作，则添加结果，否则 Spark 执行器继续向主报告结果。

这不是将工作重新并行化的好方法，它总是会给您带来痛苦，让 Spark 为您完成。

希望你明白这里发生了什么

谢谢