Spark Dataframe 过滤器优化

Question

我正在从 s3 存储桶中读取大量文件。

读完这些文件后，我想对数据框进行过滤操作。

但是当执行过滤操作时，数据会再次从 s3 存储桶中下载。如何避免数据框重新加载？

我已尝试在过滤操作之前缓存和/或持久化数据帧。但是，仍然以某种方式再次从 s3 存储桶中提取数据。

var df = spark.read.json("path_to_s3_bucket/*.json")

df.persist(StorageLevel.MEMORY_AND_DISK_SER_2)

df = df.filter("filter condition").sort(col("columnName").asc)

如果数据帧被缓存，则不应再次从 s3 重新加载。

Answer 1

当你打电话时

var df = spark.read.json("path_to_s3_bucket/*.json")

幕后发生的事情是 spark 进行分区发现、文件列表和架构推断（如果您有很多文件，这可能会在后台运行 sum 作业以并行执行文件列表）。

下次调用时

df.persist(StorageLevel.MEMORY_AND_DISK_SER_2)

只有您想要持久化数据的信息被传递给查询计划，但此时没有持久化（这是一个惰性操作）。

下次调用时

df = df.filter("filter condition").sort(col("columnName").asc)

再次只更新查询计划。

现在如果调用show()、count()等动作，就会处理查询计划并执行spark job。所以现在数据将被加载到集群上，它会被写入内存（因为缓存），然后从缓存中读回，根据您的查询计划对其进行过滤、排序和进一步处理。