斯卡拉/ Spark : Immutable Dataframes and Memory

Question

我是 Scala 的新手。我有 Java 和 R 方面的经验

我对数据帧的不变性和内存管理感到困惑。原因是这样的:

R 中的 Dataframe 也是不可变的。后来发现在R里行不通。 (简单地说)当处理大量的列时，每个转换都会产生一个新的 Dataframe。对 1000 个连续列进行 1000 次连续操作将导致 1000 个 Dataframe 对象)。现在，大多数数据科学家更喜欢 R 的 data.table，它通过引用对单个 data.table 对象执行操作。

Scala 的数据框(对新手来说)似乎有类似的问题。例如，以下代码在重命名 1000 列时似乎创建了 1000 个数据帧。尽管有 foldLeft()，每次调用 withColumn() 都会创建一个新的 DataFrame 实例。

那么，我是相信 Scala 中非常高效的垃圾收集，还是需要尝试限制创建的不可变实例的数量。如果是后者，我应该关注哪些技术？

def castAllTypedColumnsTo(df: DataFrame,
                        sourceType: DataType, targetType: DataType): 
DataFrame =
{

val columnsToBeCasted = df.schema
  .filter(s => s.dataType == sourceType)

if (columnsToBeCasted.length > 0)
{
  println(s"Found ${columnsToBeCasted.length} columns " +
    s"(${columnsToBeCasted.map(s => s.name).mkString(",")})" +
    s" - casting to ${targetType.typeName.capitalize}Type")
}

columnsToBeCasted.foldLeft(df)
{ (foldedDf, col) =>
  castColumnTo(foldedDf, col.name, targetType)
}
}

这个方法会在每次调用时返回一个新的实例

  private def castColumnTo(df: DataFrame, cn: String, tpe: DataType): 
DataFrame =
{

//println("castColumnTo")
df.withColumn(cn, df(cn).cast(tpe)

)
}

Answer 1

区别本质上是懒惰。返回的每个新 DataFrame 都不会在内存中具体化。它只存储基础 DataFrame 和应该应用于它的函数。它本质上是关于如何创建一些数据的执行计划，而不是数据本身。

当实际执行并将结果保存在某处时，所有 1000 个操作都可以并行应用于每一行，因此您会得到 1 个额外的输出 DataFrame。 Spark 将尽可能多的操作压缩在一起，并且不会实现任何不必要的或未明确请求保存或缓存的操作。