python-2.7 - 什么时候在 spark 数据帧上使用 persist() 性能不实用？

Question

在努力提高代码性能的同时，因为我有很多工作失败(中止)，我想使用 persist()每当我需要在许多其他操作中使用相同的数据帧时，在 Spark 数据帧上运行。在执行此操作并跟踪 Spark 应用程序 UI 中的各个阶段时，我觉得这样做并不总是最佳，这取决于分区数量和数据大小。我不确定直到我因为持久化阶段的失败而中止了工作。

我在质疑 如果使用 persist() 的最佳实践每当对数据帧执行许多操作时总是有效的？ 如果不是，什么时候不是？如何判断？

更具体地说，我将展示我的代码和中止作业的详细信息:

#create a dataframe from another one df_transf_1 on which I made a lot of transformations but no actions
spark_df = df_transf_1.select('user_id', 'product_id').dropDuplicates()
#persist
spark_df.persist()
products_df = spark_df[['product_id']].distinct()
df_products_indexed = products_df.rdd.map(lambda r: r.product_id).zipWithIndex().toDF(['product_id', 'product_index'])

你可能会问我为什么坚持 spark_df ?
这是因为我将多次使用它，例如 products_df还有 joins (例如: spark_df = spark_df.join(df_products_indexed,"product_id")


第 3 阶段失败原因的详细信息:

Job aborted due to stage failure: Task 40458 in stage 3.0 failed 4 times, most recent failure: Lost task 40458.3 in stage 3.0 (TID 60778, xx.xx.yyyy.com, executor 91): ExecutorLostFailure (executor 91 exited caused by one of the running tasks) Reason: Slave lost Driver stacktrace:

输入数据的大小( 4 TB )很大，在做持久化之前有没有办法检查数据的大小？它是选择坚持或不坚持的一个参数吗？还有 persist 的分区(任务)数 > 100,000

Answer 1

以下是使用 persist() 的两种情况:

使用后repartition为了避免在下一步使用数据帧时一次又一次地改组数据。这仅在您为持久化数据帧/RDD 调用多个操作的情况下有用，因为持久化是一种转换，因此被懒惰地评估。通常，如果您在同一个数据帧/RDD 上有多个操作。

迭代计算，例如当您想在 for 循环中查询数据帧时。与 persist Spark 将保存中间结果并省略对每个 Action 调用重新评估相同的操作。另一个例子是用 join 附加新列。如讨论 here .