Spark - Kryo 与 JavaSerialization。一样大吗？

Question

我在 Spark 中使用缓存。现在，我使用了几个缓存，其中一些在内存中约为 20gb。我首先尝试了 cache() ，后来又尝试了 persist 和 MEMORY_SER ，大小很大，所以我改为 java 序列化，其中一些得到了大约 20gb。 现在，我想使用 Kryo，我已经注册了类并且我没有收到任何错误，但是它的大小与我在大多数缓存中使用 Kryo 执行它时的大小相同。

我想缓存的一些对象是这样的：

case class ObjectToCache(id: Option[Long],
                      listObject1: Iterable[ObjectEnriched],
                       mp1: Map[String, ObjectEnriched2],
                       mp2: Map[String, ObjectEnriched3],
                       mp3: Map[String, ObjectEnriched4])

我在 Kryo 注册为：

kryo.register(classOf[Iterable[ObjectEnriched2]])
kryo.register(classOf[Map[String,ObjectEnriched3]])
kryo.register(classOf[Map[String,ObjectEnriched4]])
kryo.register(ObjectEnriched)
kryo.register(ObjectEnriche2)
kryo.register(ObjectEnriched3)
kryo.register(ObjectEnriched4)

我做错了吗？有什么方法可以知道它是否使用 Kryo？ 我认为它正在使用，因为在某些时候我收到了一个错误，因为我没有剩余空间：

Serialization trace:
mp1 (ObjectEnriched)
        at com.esotericsoftware.kryo.io.Output.flush(Output.java:183)
        at com.twitter.chill.Tuple2Serializer.write(TupleSerializers.scala:37)
        at com.twitter.chill.Tuple2Serializer.write(TupleSerializers.scala:33)

我正在使用带有 Spark Streaming 的 RDD。

Answer 1

检查数据帧（DF）是否被缓存 只需通过调用操作 df.show 来触发缓存，并在 http://localhost:4040/storage 检查 spark UI 以查看 DF 是否已缓存。您应该在那里看到。

您也可以使用 queryExecution 或 explain 来查看 InMemoryRelation

scala> df.queryExecution.withCachedData

res0: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan = InMemoryRelation [id#0L], true, 10000, StorageLevel(磁盘, 内存, 反序列化, 1 个副本) +- *Range (0, 1, step=1, splits=Some(8))

还可以尝试使用 Datasets 而不是 DataFrame。 DataSet 不使用标准的序列化方法。他们使用专门的列式存储和自己的压缩方法，您甚至不需要使用 Kryo 序列化器存储您的数据集。

Answer 2

我做错了吗？有什么方法可以知道它是否使用 Kryo？

您确实在使用 kryo，它正在正确序列化您的对象。

如果你设置了标志：

conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer") 
conf.set("spark.kryo.registrator","com.orange.iris.common.serializer.CustomKryoRegistrator")

那么它肯定会使用 Kryo 序列化器。另外，由于您包含标志：

conf.set("spark.kryo.registrationRequired", "true") 

如果它试图序列化一个未注册的类，它将失败（有关更多信息，请参阅此answer）。

你有多少内存？ 如果您使用 Java 和 Kryo 序列化的大小大致相同，并且您坚持使用 MEMORY_ONLY_SER，那么即使使用 Kryo 和 Spark 正在重新计算不适合的部分，您的分区仍然可能不适合内存.这将导致大小相同。

有几种方法可以解决这个问题，即运行作业并持久保存到 MEMORY_AND_DISK_SER，然后在使用 Kryo 时检查磁盘溢出。有关存储级别的更多信息，请参阅here