在转换工作流程中持久化 DataFrame答案

【问题标题】：Persisting of DataFrames in transformation workflows在转换工作流程中持久化 DataFrame
【发布时间】：2016-02-20 10:15:06
【问题描述】：

我正在尝试对 3 个 DataFrame 进行一系列转换。每次转换后，我都想保留 DF 并将其保存到文本文件中。我正在做的步骤如下。

第 0 步：
创建 DF1
创建 DF2
创建 DF3
创建 DF4
（还没有持久化没有保存）

第 1 步：
通过连接 DF1 和 DF2 创建 RESULT-DF1
将其持久化到磁盘和内存
保存到文本文件

第 2 步：
通过连接 RESULT-DF1 和 DF3 创建 RESULT-DF2
将其持久化到磁盘和内存
保存到文本文件

第 3 步：
通过连接 RESULT-DF2 和 DF4 创建 RESULT-DF3
将其持久化到磁盘和内存
保存到文本文件

观察：
Step1 创建的任务数为 601
Step2 创建的任务数为 1004（没有跳过任何内容）
Step3创建的任务数为1400（跳过400个任务）

作为不同的方法，我将上述步骤分为三个不同的运行。 IE;

开始，加载 DF1 和 DF2，执行 Step1，保存 RESULT-DF1 并退出
开始，加载 DF3，从文件加载 RESULT-DF1，执行步骤 2，保存 RESULT-DF2 并退出
开始，加载 DF4，从文件加载 RESULT-DF2，执行 Step3，保存 RESULT-DF3 并退出

后面的方法跑得更快。

我的问题是：
在第一种方法中是否遗漏了一些持久性方面的东西？
为什么 Step2 运行不只是使用 Step1 的结果，即使在持久化之后也不重做所有任务（只有 601 个任务而不是 1004 个）？
在实施此类转换工作流程时，有哪些关于最佳实践的好读物？

【问题讨论】：

您如何持久化 DataFrame 以及如何加入它们。请分享代码。
我已经尝试过这种情况并得到了这种行为的原因。您能否提供以下详细信息：1）集群内存和 2）用于创建 DF1、DF2、DF3 的文件大小
它是一个 4 + 1 集群。所有五个节点都有 32 GB。跑了四个执行者。提供 20 GB 作为执行程序内存和 20 GB 作为驱动程序内存。每个源的数据大小约为 50 到 75 GB。
@pankajarora 使用数据框 sql 连接（内部连接）。对于坚持，尝试： resDf1.persist(StorageLevel.DISK_ONLY()) 和 resDf1.persist(StorageLevel.MEMORY_AND_DISK())

标签： apache-spark

【解决方案1】：

由于没有提供代码，我假设您执行的连接操作每次都不同（在不同的属性和数据上）。即使您已经缓存了数据帧，Spark 也需要将每个连接解析为多个阶段和任务。 Catalyst 优化器负责为您的查询创建逻辑（初始和优化）和物理计划。鉴于每次必须根据您的查询和相应的数据集计算新计划时的执行顺序（每次连接后数据框可能会变小或变大）。

鉴于每次连接后任务从一个数据帧增加到另一个数据帧，您的数据集可能会变得越来越大和/或您对多个属性执行连接操作。但是，我无法理解您在第二种方法中退出的意思。

为了进一步阅读，我建议以下内容： https://databricks.com/blog/2015/04/13/deep-dive-into-spark-sqls-catalyst-optimizer.html

【讨论】：

第二种方法中的“退出”是指运行三个程序。第一个程序做一些事情，保存结果并停止。第二个程序加载第一个程序的结果，做一些事情，保存结果并停止......
你能帮忙吗？ @maasg

【解决方案2】：

在您的情况下，请考虑 Spark SQL 查询优化器的工作原理。 Catalyst 优化器目前遇到挑战的一种情况是查询计划非常大。这些查询计划往往是迭代算法的结果，如图形算法或机器学习算法。一个简单的解决方法是在每次迭代结束时将数据转换为 RDD 并返回 DataFrame/Dataset。

我遇到了与您上面描述的完全相同的问题。这种解决方法确实很有帮助。

~埃里克

【讨论】：