Richiede Kryo serializzazione a Spark (Scala)

Question

ho Kryo serializzazione accesa con questo:

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") 

voglio garantire che una classe personalizzata viene serializzato utilizzando Kryo quando mescolate tra i nodi. Posso registrare la classe con Kryo questo modo:

conf.registerKryoClasses(Array(classOf[Foo])) 

Da quanto ho capito, questo non garantisce che in realtà la serializzazione Kyro viene utilizzato; se un serializzatore non è disponibile, kryo tornerà alla serializzazione Java.

Per garantire che Kryo serializzazione accade, ho seguito questa raccomandazione dalla documentazione Spark:

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

Ma questo fa sì che IllegalArugmentException da buttare ("classe non è registrata") per un gruppo di classi diverse, che mi assumere Spark utilizza internamente, ad esempio i seguenti:

org.apache.spark.util.collection.CompactBuffer 
scala.Tuple3 

Sicuramente non devo registrare manualmente ciascuna di queste singole classi con Kryo? Questi serializzatori sono tutti definiti in kryo, quindi esiste un modo per registrarli automaticamente?

Answer 1

A quanto ho capito, questo non garantisce effettivamente la serializzazione del kyro; se un serializzatore non è disponibile, kryo tornerà alla serializzazione Java.

No. Se si imposta spark.serializer su org.apache.spark.serializer. KryoSerializer, Spark utilizzerà Kryo. Se Kryo non è disponibile, riceverai un errore. Non c'è alcun ripiego.

Quindi cos'è questa registrazione Kryo?

Quando Kryo serializza un'istanza di una classe non registrata, deve restituire il nome di classe completo. Sono molti personaggi Invece, se una classe è stata preregistrata, Kryo può semplicemente emettere un riferimento numerico a questa classe, che è solo 1-2 byte.

Ciò è particolarmente importante quando ogni riga di un RDD viene serializzata con Kryo. Non si desidera includere lo stesso nome di classe per ciascuna di un miliardo di righe. Quindi pre-registrati queste lezioni. Ma è facile dimenticare di registrare una nuova classe e poi stai sprecando di nuovo i byte. La soluzione è richiedere che ogni classe sia registrata:

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

Ora Kryo non genererà mai nomi di classi completi. Se incontra una classe non registrata, si tratta di un errore di runtime.

Sfortunatamente è difficile enumerare tutte le classi che verranno serializzate in anticipo. L'idea è che Spark registri le classi specifiche di Spark e tu registri tutto il resto. Hai un RDD[(X, Y, Z)]? Devi registrarti classOf[scala.Tuple3[_, _, _]].

Il list of classes that Spark registers include effettivamente CompactBuffer, quindi se si verifica un errore, si sta facendo qualcosa di sbagliato. Stai bypassando la procedura di registrazione Spark. Devi usare spark.kryo.classesToRegister o spark.kryo.registrator per registrare i tuoi corsi. (Vedi lo config options. Se usi GraphX, il tuo registratore dovrebbe chiamare GraphXUtils. registerKryoClasses.)

Answer 2

In base a quello che stai vedendo, ipotesi migliore è che ti manca la dichiarazione:

sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") 

Negli ultimi giorni ho anche alle prese con la conversione di serializzazione per Kryo, anche per Graphx , inclusa la registrazione di scala.Tuple3 con Kryo, apparentemente perché il codice Spark/GraphX ​​sta creando una Tuple3 quando eseguo un 'sortBy'.

Sono stati aggiunti un gruppo di altre classi, uno alla volta, da elencare per registrarsi con Kryo, per lo più con classi Scala e Spark, non credo che avrei bisogno di aggiungere. Pensando/sperando ci debba essere un modo migliore per usare Kryo con Spark.