通过缓存重用相同的数据帧

Question

我有一个名为 dataframe_1 的 DataFrame，它在我的 PySpark 代码的多个部分中被引用。为了优化性能，我打算缓存 dataframe_1 ，以便后续引用使用缓存的数据，而不是多次重新读取它。

原始代码：

dataframe_1 = spark.read.format("delta").table("schema.table_name")
dataframe_1 = dataframe_1.filter(condition).select(list_of_columns_needed)

# Planning to use cache here:dataframe_1.cache() so that cached data can be used by multiple references below.

# usage-1 of dataframe_1
columns_to_be_renamed = ["col_1","col_2","col_3"]
for c in columns_to_be_renamed:
    dataframe_1 = dataframe_1.withColumn(c, 
        when(trim(col(c)) == "", None)
        .otherwise(concat(lit(c), lit("_"), trim(col(c))))
    )

# usage-2 and other references of dataframe_1
...

我需要重命名所有列值。在我重命名 columns_to_be_renamed 中提到的列值的循环中，我需要将 DataFrame 名称维护为 dataframe_1。如果我将值分配给 dataframe_2 如下（我想要一些新的 df，以便我可以缓存旧的 df），则仅更新最后一列的值（这不是预期的行为；理想情况下，应更新所有列，如上面的代码）：

columns_to_be_renamed = ["col_1","col_2","col_3"]
for c in columns_to_be_renamed:
    dataframe_2 = dataframe_1.withColumn(c, 
        when(trim(col(c)) == "", None)
        .otherwise(concat(lit(c), lit("_"), trim(col(c))))
    )

鉴于此，我应该在读取 DataFrame dataframe_1 后立即缓存它，还是应该在循环后缓存它？我担心的是，如果我在读取后立即缓存，则循环后对 dataframe_1 的后续引用可能会引用循环中的未缓存版本而不是缓存版本。

dataframe_1 = spark.read.format("delta").table("schema.table_name")
dataframe_1 = dataframe_1.filter(condition).select(list_of_columns_needed)

# New changes
dataframe_1.cache()
dataframe_2 = dataframe_1

# usage-1 of dataframe_1
columns_to_be_renamed = ["col_1","col_2","col_3"]
for c in columns_to_be_renamed:
    dataframe_2 = dataframe_2.withColumn(c, 
        when(trim(col(c)) == "", None)
        .otherwise(concat(lit(c), lit("_"), trim(col(c))))
    )

# usage-2 and other references of dataframe_1
...

我已经验证了这种情况 - 当我们执行 dataframe_2 = dataframe_1 时，稍后更改 dataframe_2，dataframe_1 不会更新，保留初始数据/状态，如预期的那样。

这是处理它的唯一方法，还是有更好的方法来处理此类用例？

Answer 1

这是处理它的好方法，但要测试 .cache 的性能，而不是简单地为您的用例重写数据。在当前的项目中，我注意到 .cache 比简单地将数据帧写入磁盘并引用它慢 2-3 倍。

此外，不要使用 withColumn，特别是如果您要重命名大量列，它会严重影响您的性能，请直接使用 select，因为您知道字段。