Dataproc 中的 Pyspark 检查点（StackOverFlowError）答案

【问题标题】：Pyspark Checkpointing in Dataproc(StackOverFlowError)Dataproc 中的 Pyspark 检查点（StackOverFlowError）
【发布时间】：2018-07-31 04:38:36
【问题描述】：

我在使用 pyspark 持久化数据集时遇到了 stackoverflow 错误。我将整个数据帧转换为双类型，然后坚持计算统计数据，我读到检查点是stackoverflow的解决方案。但是，我无法在 dataproc 中实现它。

我正在使用 pyspark，当我检查数据帧并使用 df.isCheckpointed() 检查点时，它返回 false。但是，当我调试它时，df.rdd.is_checkpointed 说 True。包裹有什么问题/我做错了吗？
我认为 localCheckpoint 更适合我的目的（https://spark.apache.org/docs/2.3.1/api/java/org/apache/spark/rdd/RDD.html#localCheckpoint()），因为我的问题只是 DAG 深度太深，但我找不到任何用例。另外，如果我只是检查点 RDD 说它是检查点（如第一个问题），但如果我尝试 localcheckpoint，它说它不是。有人试过这个功能吗？
尝试本地独立模式后，我尝试使用 dataproc。我尝试了 hdfs 和谷歌云存储，但无论哪种方式存储都是空的，但 rdd 说它是检查点。

谁能帮我解决这个问题？提前致谢！

【问题讨论】：

标签： apache-spark pyspark google-cloud-dataproc

【解决方案1】：

如果您使用 localCheckpoint，它将写入执行程序的本地磁盘，而不是 HDFS/GCS：https://spark.apache.org/docs/2.3.1/api/java/org/apache/spark/rdd/RDD.html#localCheckpoint--。

还要注意，检查点有急切（立即检查点）和非急切（RDD 实际物化时检查点）模式。这可能会影响这些方法返回的内容。代码通常是最好的文档：https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L1615。

一般来说，请在此类问题上附上示例代码（repros）——这样我们可以更直接地回答您的问题。

【讨论】：

对于localcheckpoint，我已经读到它按照你所说的写入本地磁盘，所以我没有为此设置任何目录。我正在寻找可以检查我是否正确设置检查点的方法，因为我提到的所有标志都是错误的。
我不熟悉“物化”这个词，但是在检查点之后，我正在持久化 rdd 并收集统计信息，即使之后我在目录中也找不到 rdds
对不起，我不能分享整个代码，但是在我检查 rdds 的那一刻，我只是写了 dataframe.rdd.checkpoint()/dataframe.persist() 和在此之前，sparkcontext。 setCheckpointDir("gs://......." ) 用于检查点。

【解决方案2】：

使用您的 SparkSession 对象创建一个临时目录：

spark.sparkContext.setCheckpointDir("/tmp/checkpoints")


dataframe_name = # Any Spark Dataframe of type <pyspark.sql.dataframe.DataFrame>

此时，dataframe_name 将是 DAG，您可以将其存储为检查点，例如，

dataframe_checkpoint = dataframe_name.checkpoint()

dataframe_checkpoint 也是 <pyspark.sql.dataframe.DataFrame> 类型的 spark 数据框，但它存储查询结果而不是 DAG。

在以下情况下使用检查点：

计算耗时较长
计算链太长
依赖的 RDD 太多

【讨论】：