Apache Spark：有效地转换大型 DataFrame

Question

我有一个使用 DataFrame API 加载的 1GB csv 文件。我还实现了一个自定义的Transformer 来准备数据，以便Estimator 可以处理它。

transform 方法正在执行一些不同的操作：

• 铸造柱。
• 过滤行。
• 删除列。
• 创建对其他列应用函数的新列。

我担心此过程中的内存使用情况。如果在每次转换后将结果存储在变量中会发生什么？例如（简体）：

override def transform(dataset: Dataset[_]): DataFrame = {

    val df = dataset.withColumn("DayOfWeek", $"DayOfWeek".cast("int"))

    val df2 = df1.filter($"Diverted" === 0)

    val df3 = df2.drop(forbiddenVariables: _*)

    val df4 = df3.withColumn("DepHour", hourExtractorUdf($"DepTime"))

    val df5 = df4.select($"*", concat($"Origin", lit("-"), $"Dest").as("Route"))

    df5

}

假设我这样做是为了在一个转换和另一个转换之间进行记录。

好的。第二种选择。如果我使用var 而不是val 会怎样？

override def transform(dataset: Dataset[_]): DataFrame = {

    var df = dataset.withColumn("DayOfWeek", $"DayOfWeek".cast("int"))

    df = df.filter($"Diverted" === 0)

    df = df.drop(forbiddenVariables: _*)

    df = df.withColumn("DepHour", hourExtractorUdf($"DepTime"))

    df = df.select($"*", concat($"Origin", lit("-"), $"Dest").as("Route"))

    df

}

我想现在在整个过程中我没有在内存中加载 5 个 DataFrame。对吧？

最后，下一个选项呢，它更节省内存吗？

override def transform(dataset: Dataset[_]): DataFrame = {

    dataset.withColumn("DayOfWeek", $"DayOfWeek".cast("int"))
      .filter($"Diverted" === 0)
      .drop(forbiddenVariables: _*)
      .withColumn("DepHour", hourExtractorUdf($"DepTime"))
      .select($"*", concat($"Origin", lit("-"), $"Dest").as("Route"))

}

当然，我假设没有比其他选项的计算成本更高的选项了。

Answer 1

您的代码的所有版本都是等效的，因为它们最终会生成相同的数据帧并且不会产生副作用。似乎对 spark 的工作原理存在一些根本性的误解。 DataFrames 不包含任何数据。它们只是一个执行计划。

在学习 spark 中，我们经常讨论“转换”和“动作”之间的区别。

转换修改数据，例如filter、select、drop 以及任何其他修改数据框的方法。 “转换”做零工作，他们只是建立执行计划。

另一方面，动作实际上执行了一些可见的效果。这些是诸如保存到文件、将结果收集到驱动程序或使用 foreach 使用数据之类的事情。只有当一个动作被调用时，你的数据框才会被评估并运行转换。

1GB 的数据也非常小，如果你真的需要使用 spark 的话，也许要重新考虑。