在spark代码中多次引用DF时，是否每次都使用DAG创建数据帧（如果不使用persist）？

Question

我在一次面试中面对这个问题，操作顺序如下。

读取csv文件并创建DF。DF=火花。读取（）。格式（“csv”）。csv（“/路径/文件.csv”）

根据某些条件筛选数据。DF。过滤器（..）

执行计数操作。DF. count（）

将数据保存在文件/Table中。DF. saveAsTable（）

问题是有多少转换和操作，以及读取csv文件多少次以创建DF？

我很好奇想了解以下内容

一旦文件被读取，DF是否会保留在内存中，直到spark作业完成？如果DF没有在进一步的步骤中使用，即使它从内存中删除，我们也不介意。但是，如果它正在进一步的步骤中使用，它是否会保留在内存中，直到最后一次使用它的步骤？

每次触发每个操作时，是否都会读取文件并创建DF？为什么需要这样做？

我知道执行操作时，文件是只读的。

现在，当首先调用count（）操作时，文件已被读入内存，并构建了DF。所以我希望它能留在记忆中。但是，为什么saveAsTable（）操作会再次读取文件并使用DAG重新构建DF？

如果每次我们在代码中使用DF引用时都在重新评估DF（使用DAG），那么每次重建DF不是很耗时吗？如果我们每次都必须避免这个重建过程，并且如果我们开始保留DF，我们最终会不会占用所有内存并将一些数据溢出到磁盘并减慢火花应用程序的速度？

有人能解释一下发生了什么吗？

Answer 1

在Spark DAG中，所有操作都分为多个“阶段”。所有阶段都由“洗牌”操作分开。

阶段中的操作在处理位于单个分区中的数据时被组合在一起。

因此，每当执行“操作”（在本例中为count（）和saveAsTable（））时，它都会触发数据的重新计算，直到上一个DF。也就是说，除非缓存了前一个DF。

val df = spark.read().format("csv").csv("/path/file.csv")

df = df.filter(....)  // you are over writing the df reference

// df.cache() // use this if filter and read is to be performed only once.

val cnt = df.count()  // action - triggers the filter and read operations 

df.saveAsTable()   //  action - triggers the filter and read operations 

关于您的问题：

如果每次我们在代码中使用DF引用时都要重新评估DF（使用DAG），那么每次重建DF不是很耗时吗？

• 是的，这是一种浪费，这就是为什么如果DF被多次使用，建议缓存DF的原因

如果我们每次都必须避免这个重建过程，并且如果我们开始保留DF，我们最终会不会占用所有内存并将一些数据溢出到磁盘并减慢Spark应用程序的速度？

• 我想你已经回答了，是的，记忆是有限的资源。这就是默认情况下不会持久化所有DFs的原因。但是，当您持久化时，除了其他选项外，还有多个选项，如MEMORY\u ONLY、MEMORY\u和磁盘。在此处检查详细信息