Set a more efficient serializer

[JulienPeloton]

One needs to introduce KryoSerializer.

[JulienPeloton]

Note to myself:
https://github.com/EsotericSoftware/kryo#default-serializers

[JulienPeloton]

From my experience, just registering classes does not bring any improvement (current implementation) with Spark 2+:

/** Main App */
val conf = new SparkConf()
conf.registerKryoClasses(
  Array(classOf[MyClass1], etc...)
)

As far as I understand, Spark 2+ is already using Kryo serializer for many common things, but custom classes still require custom declaration, e.g.

/** File SpatialSerializer.scala */
import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.Input
import com.esotericsoftware.kryo.io.Output
import com.esotericsoftware.kryo.Serializer

class SpatialSerializer extends Serializer {
  // ...

  def write(Kryo kryo, Output output) {
    // ...
  }
    
  def read(Kryo kryo, Input input) {
    // ...
  }
}

and then

/** File spark3dKryoRegistrator.scala */
import com.esotericsoftware.kryo.Kryo

class spark3dKryoRegistrator implements KryoRegistrator {
  override def registerClasses(Kryo kryo) {
    // Instantiate your serializer
    val serializer = new SpatialSerializer()

    // Register your classes
    kryo.register(MyClass1.class, serializer)
    // etc...
  }
}

and finally

/** Main App */
val conf = new SparkConf()
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.set("spark.kryo.registrator", classOf[spark3dKryoRegistrator].getName)

...

That would be good to have it implemented, and benchmarked.