持久性在 Spark 中的工作原理

Posted 2023-04-17

【中文标题】持久性在 Spark 中的工作原理

【英文标题】：How persistence works in Spark

【发布时间】：2018-08-17 13:48:25

【问题描述】：

我正在持久化一些存储在 var 中的数据帧。现在，当该 var 的值发生变化时，持久性如何工作？例如：

var checkedBefore_c = AddressValidation.validateAddressInAI(inputAddressesDF, addressDimTablePath, target_o, target_c, autoSeqColName).distinct.filter(col(CommonConstants.API_QUALITY_RATING) >= minQualityThreshold)
checkedBefore_c.persist(StorageLevel.MEMORY_AND_DISK_SER)

var pre_checkedBefore_c = checkedBefore_c.except(checkedBefore_o)
pre_checkedBefore_c.persist(StorageLevel.MEMORY_AND_DISK_SER)

checkedBefore_c = pre_checkedBefore_c.drop(target_o).drop(autoSeqColName)
          .withColumn(target_o, pre_checkedBefore_c(target_c))
          .withColumn(CommonConstants.API_STATUS, lit("AI-INSERT"))
          .withColumn(CommonConstants.API_ERROR_MESSAGE, lit(""))

checkedBefore_c = CommonUtils.addAutoIncremetColumn(checkedBefore_c, autoSeqColName)
checkedBefore_c = checkedBefore_c.select(addDimWithLoggingSchema.head, addDimWithLoggingSchema.tail: _*)
checkedBefore_c.persist(StorageLevel.MEMORY_AND_DISK_SER) 

【问题讨论】：

缓存打破了沿袭。第一次致电persis 弊大于利。如果您直到最后都没有调用checkedBefore_c，Spark 可能仍会根据您在checkedBefore_c 上进行的一系列转换进行一些优化。

因此，对于一个动作，如果我在 checked_before_c 上编写 n 个过滤器以生成 n 个新数据帧，然后我将加入所有这些新数据帧以形成一个我执行写入的数据帧。不需要持久性。我说的对吗？

【参考方案1】：

您正在尝试保留 checkedBefore_c DataFrame，但在您的代码中您没有调用任何操作。

简要说明

Spark 有两种操作，转换和动作。

转换：转换是惰性的，例如 map、reduceByKey 等。

动作：动作是急切的，例如 foreach、count、save 等。

persist和cache也是惰性操作，所以直到你调用action的时候persist和cache都不会被执行。

更多细节请参考 Spark 中的动作。您也可以参考this。

现在坚持是如何工作的。 在坚持中，火花存储分区在内存或磁盘或两者中。 它们是各种选项，所有选项参考 org.apache.spark.storage.StorageLevel 源代码。

每个 executors 将负责存储他们的分区，如果给出了内存选项，首先它将尝试适合所有分区，如果它不适合，然后它将基于旧的缓存数据清理（它是 LRY 缓存）。如果仍然所有分区都不适合内存，它将缓存适合内存的分区并保留其余分区。

如果选择了带磁盘的内存选项，则首先执行上述所有步骤，然后将左分区存储在本地磁盘中。

如果复制因子为 2，则每个分区将缓存在两个不同的执行器中。

在您的情况下，您已经传递了 MEMORY_AND_DISK_SER，这意味着所有对象都将在缓存之前被序列化。默认情况下使用 Java 序列化，但您可以覆盖它并使用推荐的 Kyro 序列化。

【讨论】：

是的。我在我的代码中使用了动作，你对惰性评估是正确的。但我的问题是“我已经坚持了两次checkBefore_c，因为在第一次坚持之后，我用新的df改变了变量的值”。是否需要第二次坚持它。如果不是，是否每次我们更改值时都会保留更新的数据帧

您需要第二次持久化它，因为它不会自动持久化更新的数据帧。

我认为第一次持久性不是必需的。在优化方面是否需要？

我没有看到任何优化。简而言之，在两个不同的动作中使用或在迭代算法中使用的 RDD/DataFrame 是缓存的良好候选者。在这里，您既不是在不同的动作中使用它，也不是在做迭代算法

好的。这对我会有帮助。因此，对于一项操作，如果我在 checked_before_c 上编写 n 个过滤器以生成 n 个新数据帧，然后我将加入所有这些新数据帧以形成一个我执行写入的数据帧。不需要持久性。我说的对吗？