Как я могу оценить размер в байтах каждого столбца в кадре данных Spark?

Question

У меня очень большой Spark DataFrame с несколькими столбцами, и я хочу сделать обоснованное суждение о том, сохранять или нет их в моем конвейере, отчасти исходя из того, насколько они велики.Под «насколько большим» я подразумеваю размер в байтах в оперативной памяти, когда этот DataFrame кэшируется, что, я думаю, будет достойной оценкой вычислительных затрат на обработку этих данных.Некоторые столбцы являются простыми типами (например, double, integer), но другие являются сложными типами (например, массивы и карты переменной длины).

Подход, который я пробовал, заключается в кэшировании DataFrame без, а затем с соответствующим столбцом, откройте вкладку «Хранилище» в интерфейсе Spark и возьмите разницу.Но это раздражающее и медленное упражнение для DataFrame с большим количеством столбцов.

Обычно я использую PySpark, поэтому предпочтительным является ответ PySpark, но Scala также подойдет.

Answer 1

Я нашел решение, основанное на следующем ответе: 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 байтов в строке.