时间戳

首页 > 解决方案 > 如何估计 Spark DataFrame 中每列的字节大小？

apache-spark - 如何估计 Spark DataFrame 中每列的字节大小？

问题描述

我有一个非常大的包含许多列的 Spark DataFrame，我想就是否将它们保留在我的管道中做出明智的判断，部分基于它们的大小。“多大”是指缓存此 DataFrame 时 RAM 中的字节大小，我希望这是对处理此数据的计算成本的一个不错的估计。一些列是简单类型（例如双精度、整数），但其他列是复杂类型（例如可变长度的数组和映射）。

我尝试过的一种方法是先缓存 DataFrame，然后再缓存有问题的列，查看 Spark UI 中的 Storage 选项卡，然后获取差异。但对于具有大量列的 DataFrame，这是一个烦人且缓慢的练习。

我通常使用 PySpark，所以 PySpark 答案会更好，但 Scala 也可以。

标签： apache-sparkpyspark

解决方案

我找到了一个基于此相关答案的解决方案：https ://stackoverflow.com/a/49529028 。

假设我正在使用一个名为的数据框df和一个名为的SparkSession对象spark：

import org.apache.spark.sql.{functions => F}

// force the full dataframe into memory (could specify persistence
// mechanism here to ensure that it's really being cached in RAM)
df.cache()
df.count()

// calculate size of full dataframe
val catalystPlan = df.queryExecution.logical
val dfSizeBytes = spark.sessionState.executePlan(catalystPlan).optimizedPlan.stats.sizeInBytes

for (col <- df.columns) {
    println("Working on " + col)

    // select all columns except this one:
    val subDf = df.select(df.columns.filter(_ != col).map(F.col): _*)

    // force subDf into RAM
    subDf.cache()
    subDf.count()

    // calculate size of subDf
    val catalystPlan = subDf.queryExecution.logical
    val subDfSizeBytes = spark.sessionState.executePlan(catalystPlan).optimizedPlan.stats.sizeInBytes

    // size of this column as a fraction of full dataframe
    val colSizeFrac = (dfSizeBytes - subDfSizeBytes).toDouble / dfSizeBytes.toDouble
    println("Column space fraction is " + colSizeFrac * 100.0 + "%")
    subDf.unpersist()
}

一些确认这种方法给出了合理的结果：

  1. 报告的列大小加起来为 100%。
  2. 简单类型的列，如整数或双精度数，每行占用预期的 4 个字节或 8 个字节。

推荐阅读