不幸的是，我无法从SizeEstimator中获得可靠的估计值，但我可以找到另一种策略-如果缓存了该帧，我们可以从queryExecution中提取其大小，如下所示：

df.cache.foreach(_ => ())
val catalyst_plan = df.queryExecution.logical
val df_size_in_bytes = spark.sessionState.executePlan(
    catalyst_plan).optimizedPlan.stats.sizeInBytes

对于示例的Parquet rame，这正好是4.8GB（这也对应于写入未压缩的Parquet表时的文件大小）。
这有一个缺点，即需要缓存嵌套框架，但在我的情况下，这不是一个问题。
编辑：将df.cache.foreach(_=>_)替换为df.cache.foreach(_ => ())，感谢@DavidBenedeki在评论中指出。

SizeEstimator返回一个对象在JVM堆上占用的字节数。这包括对象引用的对象，实际对象大小几乎总是要小得多。
您所观察到的大小差异是因为在JVM上创建新对象时，引用也会占用内存，这正在被计算在内。
点击这里查看文档
https://spark.apache.org/docs/2.2.0/api/scala/index.html#org.apache.spark.util.SizeEstimator$

除了尺寸估计器，你已经尝试过了（好的洞察力）。
下面是另一个选项

RDDInfo[] getRDDStorageInfo()

返回关于哪些RDD被缓存，它们是在缓存中还是在两者上，它们占用多少空间等的信息。
实际上spark存储选项卡使用了这个。Spark文档
下面是implementation from spark

/**
   * :: DeveloperApi ::
   * Return information about what RDDs are cached, if they are in mem or on disk, how much space
   * they take, etc.
   */
  @DeveloperApi
  def getRDDStorageInfo: Array[RDDInfo] = {
    getRDDStorageInfo(_ => true)
  }

  private[spark] def getRDDStorageInfo(filter: RDD[_] => Boolean): Array[RDDInfo] = {
    assertNotStopped()
    val rddInfos = persistentRdds.values.filter(filter).map(RDDInfo.fromRdd).toArray
    rddInfos.foreach { rddInfo =>
      val rddId = rddInfo.id
      val rddStorageInfo = statusStore.asOption(statusStore.rdd(rddId))
      rddInfo.numCachedPartitions = rddStorageInfo.map(_.numCachedPartitions).getOrElse(0)
      rddInfo.memSize = rddStorageInfo.map(_.memoryUsed).getOrElse(0L)
      rddInfo.diskSize = rddStorageInfo.map(_.diskUsed).getOrElse(0L)
    }
    rddInfos.filter(_.isCached)
  }

来自RDD的yourRDD.toDebugString也使用了这个。代码here

一般说明：

在我看来，要获得每个分区中记录的最佳数量，并检查您的重新分区是否正确，并且它们是否均匀分布，我建议尝试如下.

yourdf.rdd.mapPartitionsWithIndex{case (index,rows) => Iterator((index,rows.size))}
  .toDF("PartitionNumber","NumberOfRecordsPerPartition")
  .show

或现有的spark功能（基于spark版本）

import org.apache.spark.sql.functions._ 

df.withColumn("partitionId", sparkPartitionId()).groupBy("partitionId").count.show

我在这里发表评论是因为我没有足够的代表来评论公认的答案。
从Spark核心的this commit开始，方法executionPlan需要两个参数，logicalPlan和mode。
添加到hyriu的答案，对于PySpark：

df.cache().foreach(lambda x: x)
spark._jsparkSession.sessionState() \
    .executePlan(
        df._jdf.queryExecution().logical(),
        df._jdf.queryExecution().mode()   
    ).optimizedPlan() \
    .stats() \
    .sizeInBytes()

我的建议是

from sys import getsizeof

def compare_size_two_object(one, two):
    '''compare size of two files in bites'''
    print(getsizeof(one), 'versus', getsizeof(two))