Как рассчитать размер кадра данных в спарк scala

Question

Я хочу написать один большой размер кадра данных с перераспределением, поэтому я хочу вычислить количество перераспределений для моего исходного кадра данных.

numberofpartition = {размер dataframe / default_blocksize}

, поэтому, пожалуйста, сообщите мне, как рассчитать размер кадра данных в спарк scala

Заранее спасибо.

Answer 1

Используя это - spark.sessionState.executePlan (df.queryExecution.logical) .optimizedPlan.stats (spark.sessionState.conf) .sizeInBytes, мы можем получить размер фактического Dataframe после его загрузки в память, например, вы можете проверить код ниже.

scala> val df = spark.read.format("orc").load("/tmp/srinivas/")
df: org.apache.spark.sql.DataFrame = [channelGrouping: string, clientId: string ... 75 more fields]

scala> import org.apache.commons.io.FileUtils
import org.apache.commons.io.FileUtils

scala> val bytes = spark.sessionState.executePlan(df.queryExecution.logical).optimizedPlan.stats(spark.sessionState.conf).sizeInBytes
bytes: BigInt = 763275709

scala> FileUtils.byteCountToDisplaySize(bytes.toLong)
res5: String = 727 MB

scala> import sys.process._
import sys.process._

scala> "hdfs dfs -ls -h /tmp/srinivas/".!
Found 2 items
-rw-r-----   3 svcmxns hdfs          0 2020-04-20 01:46 /tmp/srinivas/_SUCCESS
-rw-r-----   3 svcmxns hdfs    727.4 M 2020-04-20 01:46 /tmp/srinivas/part-00000-9d0b72ea-f617-4092-ae27-d36400c17917-c000.snappy.orc
res6: Int = 0

val bytes = spark.sessionState.executePlan(df.queryExecution.logical).optimizedPlan.stats(spark.sessionState.conf).sizeInBytes
    val dataSize = bytes.toLong
    val numPartitions = (bytes.toLong./(1024.0)./(1024.0)./(10240)).ceil.toInt // May be you can change or modify this to get required partitions.

    df.repartition(if(numPartitions == 0) 1 else numPartitions)
      .[...]

Редактировать - 1: Пожалуйста, используйте ниже logi c в соответствии с вашими версиями свечей.

spark 2.4

val bytes = spark.sessionState.executePlan(df.queryExecution.logical).optimizedPlan.stats(spark.sessionState.conf).sizeInBytes

искра 2,3

val bytes = spark.sessionState.executePlan(df.queryExecution.logical).optimizedPlan.stats.sizeInBytes