如何强制 Spark 执行代码？答案

【问题标题】：How can I force Spark to execute code?如何强制 Spark 执行代码？
【发布时间】：2019-08-11 08:05:52
【问题描述】：

如何强制 Spark 执行对 map 的调用，即使它认为由于其惰性求值而无需执行？

我尝试将cache() 与 map 调用一起使用，但这仍然无法解决问题。我的 map 方法实际上将结果上传到 HDFS。所以，它不是没用的，但 Spark 认为它是。

【问题讨论】：

这个问题和your previous one有同样的问题：没有代码可以使用。请发布示例代码。此外，如果这与同一问题有关，请不要提出新问题。
这是一个普遍的问题。基本上，如何阻止 Spark 做出假设并执行我给它的任何代码。
如果没有您声称它做出假设的代码，我们无法判断 Spark 做出了什么假设。请发布代码。
只是出于兴趣，你为什么要这个？ Spark 是具有明确理念的 Spark。

标签： java scala hadoop apache-spark

【解决方案1】：

简答：

要强制 Spark 执行转换，您需要一个结果。有时一个简单的count 操作就足够了。

TL;DR：

好的，让我们回顾一下 RDD 操作。

RDDs 支持两种操作：

转换 - 从现有数据集创建新数据集。
actions - 在数据集上运行计算后返回一个值给驱动程序。

例如，map 是一个转换，它将每个数据集元素传递给一个函数并返回一个表示结果的新 RDD。另一方面，reduce 是一个动作，它使用某个函数聚合 RDD 的所有元素并将最终结果返回给驱动程序（尽管也有一个并行的 reduceByKey 返回分布式数据集）。

Spark 中的所有转换都是惰性的，因为它们不会立即计算结果。

相反，他们只记得应用于某些基础数据集（例如文件）的转换。 仅当操作需要将结果返回给驱动程序时，才会计算转换。这种设计使 Spark 能够更高效地运行——例如，我们可以意识到通过 map 创建的数据集将在 reduce 中使用，并且仅将 reduce 的结果返回给驱动程序，而不是更大的映射数据集。

默认情况下，每个转换后的RDD 可能会在您每次对其运行操作时重新计算。但是，您也可以使用persist（或cache）方法将RDD 持久化到内存中，在这种情况下，Spark 会将元素保留在集群中，以便下次查询时更快地访问它。还支持在磁盘上持久化RDDs，或跨多个节点复制。

结论

要强制 Spark 执行对 map 的调用，您需要一个结果。有时count 操作就足够了。

参考

Spark Programming Guide。

【讨论】：

持久化和缓存有什么区别？如果只使用一次，RDD 是否仍然存在于内存中？
使用cache，您只使用默认存储级别MEMORY_ONLY。使用persist，您可以指定所需的存储级别。如果您想为 RDD 分配除 MEMORY_ONLY 之外的另一个存储级别，请使用 persist
使用“take”触发persist怎么样？我用 1.6.1 版本做了一个实验，“count”需要比“take”多一个阶段（由 shuffle 和聚合组成）。所以我认为使用“采取”行动更有效。
某处是否有所有操作的列表？
@rsmith54 spark.apache.org/docs/2.1.1/programming-guide.html#actions 给出了最常见的那些，并且应该有一个指向文档的链接，以获取您使用的任何语言的详尽列表

【解决方案2】：

Spark transformations 只描述必须做什么。要触发执行，您需要action。

在您的情况下，存在更深层次的问题。如果目标是产生某种副作用，比如在 HDFS 上存储数据，那么正确的使用方法是 foreach。它既是一个动作，又具有清晰的语义。同样重要的是，与map 不同的是，它并不意味着引用透明。

【讨论】：

foreach 也会并行执行吗？能举个例子吗？
是的，它在工作节点上并行执行。最简单的事情是记录或打印东西。使用 PySpark：from __future__ import print_function; rdd.foreach(print)。另一种选择是foreachPartition。