apache-spark - Spark createDataFrame(df.rdd, df.schema) 与 checkPoint 打破沿袭

Question

我目前正在使用

val df=longLineageCalculation(....)
val newDf=sparkSession.createDataFrame(df.rdd, df.schema)
newDf.join......

为了在计算计划时节省时间，但是文档说检查点是“剪切”沿袭的建议方法。但我不想付出将 RDD 保存到磁盘的代价.

我的进程是一个批处理进程，时间不太长，可以毫无问题地重新启动，因此检查点对我来说没有好处(我认为)。

使用“我的”方法可能会出现哪些问题？ (文档建议使用检查点，这是更昂贵的，而不是破坏谱系的检查点，我想知道原因)

我唯一能猜测的是，如果某个节点在我的“谱系破坏”之后失败，也许我的进程会失败，而检查点节点会正常工作？ (如果 DF 被缓存而不是检查点会怎样？)

谢谢!

编辑:

From SMaZ answer, my own knowledge and the article which he provided. Using createDataframe (which is a Dev-API, so use at "my"/your own risk) will keep the lineage in memory (not a problem for me since I don't have memory problems and the lineage is not big).

With this, it looks (not tested 100%) that Spark should be able to rebuild whatever is needed if it fails.

As I'm not using the data in the following executions, I'll go withcache+createDataframe versus checkpointing (which If i'm not wrong, isactually cache+saveToHDFS+"createDataFrame").

My process is not that critical (if it crashes) since an user will be always expecting the result and they launch it manually, so if it gives problems, they can relaunch (+Spark will relaunch it) or call me, so I can take some risk anyways, but I'm 99% sure there's no risk :)

Answer 1

让我开始使用以下行创建数据框:

val newDf=sparkSession.createDataFrame(df.rdd, df.schema)

如果我们仔细研究SparkSession类，则该方法用 @DeveloperApi 进行注释。要了解此注释的含义，请查看 DeveloperApi 中的以下几行类

A lower-level, unstable API intended for developers.

Developer API's might change or be removed in minor versions of Spark.

因此，不建议将此方法用于生产解决方案，在开源世界中称为使用风险自负实现。

但是，让我们更深入地了解当我们从 RDD 调用 createDataframe 时会发生什么。它正在调用internalCreateDataFrame私有(private)方法并创建LogicalRDD。

LogicalRDD 在以下情况下创建:

• 向检查点请求数据集
• SparkSession 被请求从内部二进制行的 RDD 创建 DataFrame

因此，它与检查点操作相同，无需物理保存数据集。它只是从内部二进制行和架构的 RDD 创建 DataFrame。这可能会截断内存中的谱系，但不会截断物理级别。

所以我相信这只是创建另一个 RDD 的开销，并且不能用作检查点的替代品。

现在，检查点是截断谱系图并将其保存到可靠的分布式/本地文件系统的过程。

为什么选择检查点？

• 如果计算需要很长时间或谱系太长或取决于太多的RDD

• 保留大量沿袭信息会带来内存成本。

• 即使 Spark 应用程序终止，检查点文件也不会自动删除，因此我们可以将其用于其他进程

What are the problems which can arise using "my" method? (Docs suggests checkpointing, which is more expensive, instead of this one for breaking lineages and I would like to know the reason)

This文章将提供有关缓存和检查点的详细信息。 IIUC，你的问题更多的是我们应该在哪里使用检查点。让我们讨论一些检查点有用的实际场景

1. 让我们假设一个场景，我们有一个数据集，我们想要对其执行 100 次迭代操作，并且每次迭代都将最后一次迭代结果作为输入(Spark MLlib 用例)。现在，在这个迭代过程中，谱系将在此期间不断增长。这里定期检查数据集(假设每 10 次迭代)将确保在发生任何故障时我们可以从最后一个故障点开始该过程。
2. 让我们举一些批处理示例。想象一下，我们有一批正在创建一个具有繁重谱系或复杂计算的主数据集。现在，经过一些定期的间隔后，我们得到了一些应该使用之前计算的主数据集的数据。在这里，如果我们检查主数据集，那么它可以被不同 sparkSession 的所有后续进程重用。

My process is a batch process which is not-so-long and can be restarted without issues, so checkpointing is not benefit for me (I think).

这是正确的，如果您的进程不是重计算/大沿袭，那么就没有检查点的意义。 经验法则是，如果您的数据集没有多次使用，并且重建速度快于检查点/缓存所花费的时间和资源，那么我们应该避免它。它将为您的流程提供更多资源。