持久性在 Spark 中的工作原理

Question

我正在持久化一些存储在 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) 

Answer 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 序列化。