apache-spark - 如何找到 RDD 的大小

Question

我有RDD[Row]，需要将其保存到第三方存储库。但此第三方存储库在一次调用中最多接受 5 MB。

所以我想根据 RDD 中存在的数据大小而不是 RDD 中存在的行数来创建分区。

如何找到RDD的大小并基于它创建分区？

Answer 1

正如 Justin 和 Wang 提到的，获取 RDD 的大小并不简单。我们可以做一个估计。

我们可以对 RDD 进行采样，然后使用 SizeEstimator以获得样本的大小。正如王和贾斯汀提到的，根据离线采样的大小数据，假设 X 行离线使用了 Y GB，Z 行在运行时可能需要 Z*Y/X GB

这里是获取 RDD 大小/估计的示例 scala 代码。

我是 Scala 和 Spark 的新手。下面的示例可能会以更好的方式编写

def getTotalSize(rdd: RDD[Row]): Long = {
  // This can be a parameter
  val NO_OF_SAMPLE_ROWS = 10l;
  val totalRows = rdd.count();
  var totalSize = 0l
  if (totalRows > NO_OF_SAMPLE_ROWS) {
    val sampleRDD = rdd.sample(true, NO_OF_SAMPLE_ROWS)
    val sampleRDDSize = getRDDSize(sampleRDD)
    totalSize = sampleRDDSize.*(totalRows)./(NO_OF_SAMPLE_ROWS)
  } else {
    // As the RDD is smaller than sample rows count, we can just calculate the total RDD size
    totalSize = getRDDSize(rdd)
  }

  totalSize
}

def getRDDSize(rdd: RDD[Row]) : Long = {
    var rddSize = 0l
    val rows = rdd.collect()
    for (i <- 0 until rows.length) {
       rddSize += SizeEstimator.estimate(rows.apply(i).toSeq.map { value => value.asInstanceOf[AnyRef] })
    }

    rddSize
}