apache-spark - 与同等结构相比，为什么 Spark Row 对象如此之大？

Question

我一直在使用 java-sizeof 库 ( https://github.com/phatak-dev/java-sizeof ) 并使用它来测量 Apache Spark 中的数据集大小。事实证明，Row 对象大得离谱。就像非常大——这是为什么？

采用一个相当简单的模式:

root
 |-- account: string (nullable = true)
 |-- date: long (nullable = true)
 |-- dialed: string (nullable = true)
 |-- duration: double (nullable = true)

示例数据如下所示:

+-------+-------------+----------+--------+
|account|         date|    dialed|duration|
+-------+-------------+----------+--------+
|   5497|1434620384003|9075112643|   790.0|
+-------+-------------+----------+--------+

所以现在我们这样做:

val row = df.take(1)(0)
// row: org.apache.spark.sql.Row = [5497,1434620384003,9075112643,790.0]

所以现在我使用SizeEstimator

SizeEstimator.estimate(row)
// res19: Long = 85050896

81兆字节!对于单行!我认为这是某种错误，我这样做:

SizeEstimator.estimate(df.take(100))
// res20: Long = 85072696

有趣的是，尽管保存的数据量是原来的 100 倍，但它并没有变大多少，仅变大了 20k 左右。超过 100，它似乎变得线性。对于 1,000 行，它看起来像这样:

SizeEstimator.estimate(df.take(1000))
// res21: Long = 850711696

好的，这大约是 100 行的 10 倍——或多或少是线性的。根据测试，它以线性方式增加，持续超过 100 行。根据这些测试，大约 100 行之后，每个 Row 对象的成本仍然超过 800 KB!!

出于好奇，我为相同的基础数据尝试了几种不同的对象类型。例如，以下是 Array 对象(而不是 Row 对象)的 Array 的结果:

SizeEstimator.estimate(
  df.map(r => (r.getString(0), r.getLong(1), r.getString(2), r.getDouble(3))).take(1)
)
// res22: Long = 216

好吧，这样好一点了。更好的是，对于 10 行，只有 1976 字节，对于 100 行，只有 19,616 字节。绝对是朝着正确的方向前进。

然后，我将相同的 DataFrame 编码为 RDD[Array[Byte]]，其中每个 Array[Byte] 都是一个二进制 -编码的 Avro 记录，与底层 DataFrame 具有相同的架构。然后我这样做:

SizeEstimator.estimate(encodedRdd.take(1))
// res23: Long = 72

72 字节——更好!而且，对于 100 行，它是 5,216 字节——每行大约 52 字节，并且从那里开始不断下降(1,000 条记录为 48,656 字节)。

因此，在最好的情况下，Row 对象每 Row 的重量为 850k，而相同数据的二进制 Avro 记录约为 50 个字节.

这是怎么回事？

Answer 1

实际上Row本身并没有那么大。这就是为什么当您获取更多行时，您不会看到 as 大小发生显着变化。问题似乎是架构信息:

1. 当您收集数据时，您实际上会得到GenericRowWithSchema

   val df = Seq((1, "foo"), (2, "bar")).toDF
   df.first.getClass
   
   // res12: Class[_ <: org.apache.spark.sql.Row] = 
   //   class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
   

2. GenericRowWithSchema carries schema information来自 schema 参数:

   class GenericRowWithSchema(values: Array[Any], 
     override val schema: StructType)
   

3. 让我们确认这确实是问题的根源:

   import com.madhukaraphatak.sizeof.SizeEstimator
   import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
   
   val rowWithSchema = df.first 
   val rowWithoutSchema = new GenericRowWithSchema(
     rowWithSchema.toSeq.toArray, null)
   
   SizeEstimator.estimate(rowWithSchema)
   // Long = 1444255708
   
   SizeEstimator.estimate(rowWithoutSchema)
   // Long = 120
   

4. 假设:您看到的估计大小包括架构的大小:

   SizeEstimator.estimate(df.schema)
   // Long = 1444361928
   

   与收集的行的数量级大致相同。让我们从头开始创建一个新架构:

   import org.apache.spark.sql.types._
   
   val schema = StructType(Seq(
     StructField("_1",IntegerType,false), 
     StructField("_2",StringType,true)))
   
   
   val anotherRowWithSchema = new GenericRowWithSchema(
     Array(0, "foo"), schema) 
   
   SizeEstimator.estimate(anotherRowWithSchema)
   // Long = 1444905324
   

   如您所见，结果是一致的。

5. 为什么模式这么大？很难说。当您查看代码时，您会看到 StructType是一个复杂的类，甚至不包括其伴生对象，而不是一个简单的模式定义。

   但它没有解释报告的大小。我怀疑这可能是 SizeEstimator 中的一些侥幸，但我还不确定。

6. 您可以进一步隔离问题，但估计单个 StructField 的大小:

   import org.apache.spark.sql.types._
   import com.madhukaraphatak.sizeof.SizeEstimator
   
   object App {
     def main(args: Array[String]) {
       val schema = StructField("foo", IntegerType, true)
       println(SizeEstimator.estimate(schema))
       // 271872172
     }
   }