背景

Spark 3.5
最近在看Spark UI 上的一些指标看到一个很有意思的东西, 相邻的Shuffle Exechange 和 BroadcastExechange 中的 datasize 居然不一样，
前者为 765KB, 后者为 64.5MB。差别还不少，中间就增加了一个 AQEShuffleRead 计划

结论

Shuffle Exechange 中的是真实 UnsafeRow的大小
BroadcastExechange 中的是 MemoryBlock 类型数据结构所占的大小 ，而不是UnsafeRow的大小。
且BroadcastExechange中的datasize大小 和 2的整数倍接近。

现象以及分析

上图：

两个同样的 ShuffleExechange 记录条数和 ShuffleExechange 中 datasize 大小不一样，而在BroadcastExechange 中 dataSize 大小却是一样的（都是64.5MB）
关于 ShuffleExchange中的 dataSize的计算可以参考:Spark UI中Shuffle dataSize 和shuffle bytes written 指标区别，这里重点分析一下后者.
直接看BroadcastExechange代码:

  override lazy val relationFuture: Future[broadcast.Broadcast[Any]] = {
    SQLExecution.withThreadLocalCaptured[broadcast.Broadcast[Any]](
      session, BroadcastExchangeExec.executionContext) {
          try {
            // Setup a job tag here so later it may get cancelled by tag if necessary.
            sparkContext.addJobTag(jobTag)
            sparkContext.setInterruptOnCancel(true)
            val beforeCollect = System.nanoTime()
            // Use executeCollect/executeCollectIterator to avoid conversion to Scala types
            val (numRows, input) = child.executeCollectIterator()
            ...
            val relation = mode.transform(input, Some(numRows))
            val dataSize = relation match {
              case map: HashedRelation =>
                map.estimatedSize
              case arr: Array[InternalRow] =>
                arr.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
              case _ =>
                throw new SparkException("[BUG] BroadcastMode.transform returned unexpected " +
                  s"type: ${relation.getClass.getName}")
            }
            longMetric("dataSize") += dataSize

其中child.executeCollectIterator() 是在把数据从各个 Executor 收集到 Driver 端来，便于进行广播操作。
最主要的是 mode.transform(input, Some(numRows))，这里的数据流如下:

HashedRelationBroadcastMode.transform
  ||
  \/
HashedRelation.apply(rows, key, numRows.toInt, isNullAware = isNullAware)
  ||
  \/
UnsafeHashedRelation.apply(input, key, sizeEstimate, mm, isNullAware, allowsNullKey,
        ignoresDuplicatedKey)
  ||
  \/
new UnsafeHashedRelation(key.size, numFields, binaryMap)

最终调用的 UnsafeHashedRelation.estimatedSize的方法：

  override def estimatedSize: Long = binaryMap.getTotalMemoryConsumption
    

而 getTotalMemoryConsumption 是dataPages所占用的大小再加上longArray的大小：

  public long getTotalMemoryConsumption() {
    long totalDataPagesSize = 0L;
    for (MemoryBlock dataPage : dataPages) {
      totalDataPagesSize += dataPage.size();
    }
    return totalDataPagesSize + ((longArray != null) ? longArray.memoryBlock().size() : 0L);
  }

那么 BytesToBytesMap 是怎么分配的呢？如下：

    val binaryMap = new BytesToBytesMap(
      taskMemoryManager,
      // Only 70% of the slots can be used before growing, more capacity help to reduce collision
      (sizeEstimate * 1.5 + 1).toInt,
      pageSizeBytes)

默认的PageSize值为：defaultPageSizeBytes:

  private lazy val defaultPageSizeBytes = {
    val minPageSize = 1L * 1024 * 1024   // 1MB
    val maxPageSize = 64L * minPageSize  // 64MB
    val cores = if (numCores > 0) numCores else Runtime.getRuntime.availableProcessors()
    // Because of rounding to next power of 2, we may have safetyFactor as 8 in worst case
    val safetyFactor = 16
    val maxTungstenMemory: Long = tungstenMemoryMode match {
      case MemoryMode.ON_HEAP => onHeapExecutionMemoryPool.poolSize
      case MemoryMode.OFF_HEAP => offHeapExecutionMemoryPool.poolSize
    }
    val size = ByteArrayMethods.nextPowerOf2(maxTungstenMemory / cores / safetyFactor)
    val chosenPageSize = math.min(maxPageSize, math.max(minPageSize, size))
    if (Utils.isG1GC && tungstenMemoryMode == MemoryMode.ON_HEAP) {
      chosenPageSize - Platform.LONG_ARRAY_OFFSET
    } else {
      chosenPageSize
    }
  }

这个跟内存以及core有关。
当在进行val loc = binaryMap.lookup 以及loc.append操作的时候就会进行dataPage以及longArray的分配。而该size的大小并不是实际占用的大小，而是分配给该dataPage的大小。其实你会发现该datasize的大小几乎和2的倍数接近。