scala - Spark 多个数据帧保存

Question

我有一个 spark 作业，它创建一个数据框，我将它保存到 HDFS。我想做的是将该数据框的一个子集保存到另一个地方，但我想在这方面表现出色。

我拥有的唯一转换是保存本身……spark 作业的所有其他代码元素都是一个操作。我不缓存数据框。我担心从旧数据框创建新数据框的放置操作会再次经历所有原始数据框转换。

例如，我有这样的东西:

val df = hiveContext.read.json("hdfs://HOSTNAME:PORT/user/spark/data/in/*")

val df2 = df.withColumn("new_column", some_udf("old_column")).drop("old_column")
.
.
.

val final_df = df10.withColumn("newest_column", another_udf("old_column2")).drop("old_column2")

val subset_df = final_df.drop("this_column")
                        .drop("that_column")
                        .drop("another_column)

final_df.write.mode(SaveMode.Overwrite).format("json").save(hdfs_dir)
subset_df.write.mode(SaveMode.Overwrite).format("json").save(hdfs_dir2)

但是让我们假设 some_udf 实际上是计算密集型的。我不希望它运行两次。因此我的问题是:

我是否应该在声明 subset_df 并调用保存之前 final_df.cache() 以确保它不会再次执行 udf 的转换？

类似于:

val df = hiveContext.read.json("hdfs://HOSTNAME:PORT/user/spark/data/in/*")

val df2 = df.withColumn("new_column", some_udf("old_column")).drop("old_column")
.
.
.

val final_df = df10.withColumn("newest_column", another_udf("old_column2")).drop("old_column2")

val subset_df = final_df.drop("this_column")
                        .drop("that_column")
                        .drop("another_column)

final_df.cache()  // This is the only new line

final_df.write.mode(SaveMode.Overwrite).format("json").save(hdfs_dir)
subset_df.write.mode(SaveMode.Overwrite).format("json").save(hdfs_dir2)

Answer 1

你应该缓存:

val final_df = df10.withColumn(...)
val subset_df = final_df.drop(...)
final_df.cache() 

在第一个 Action 之前，否则它将执行两次(正如您所怀疑的那样)。