Add Spark custom Kryo registrator

[jzhuge]

Kryo can't handle guava ImmutableList and ImmutableMap. This commit add a Spark custom Kryo registrator for these classes used by GenericDataFile.

Unit test TestKryoSerialization adapted from the one in #546.

A downstream project works fine after adding this Spark conf:

spark.kryo.registrator                org.apache.iceberg.spark.SparkKryoRegistrator

Fixes #446.

This PR replaces #546 that may be merged first before the release.

[rdblue]

I think this should be included in iceberg-spark-runtime, which means that we will need to make sure we add it to the shaded LICENSE and NOTICE.

[jzhuge]

Added additional LICENSE and NOTICE.

[rdblue]

Can we add the registration to Spark's KryoSerializer from IcebergSource? It would be nice if users didn't have to add config properties.

[jzhuge]

Like the idea. Let me explore. Might have to instantiate the global SparkEnv.

[jzhuge]

Is there a way to substitute kryo with kryo-shaded?
I took a quick stab at gradle dependency substitution but couldn't get it going because --write-locks still output kryo, not kryo-shaded.

[jzhuge]

Rebase after merging #546.

TODO:

• Substitute dependency kryo with kryo-shaded
• Explore registering Kryo serializers in IcebergSource thus no need to add Spark conf spark.kryo.registrator.

[aokolnychyi]

This PR will also unblock #553 as we need to handle SerializableConfiguration in a special way when broadcasting.

[jzhuge]

Spark KryoSerializer calls Twitter's com.twitter.chill.AllScalaRegistrar that includes serializers for many Java unmodifiable collections.

[rdblue]

I think we want to avoid requiring a custom registrator for Spark to work with Kryo. Instead, we will need to use non-guava collections in places that may be serialized with Kryo.

@jzhuge, @aokolnychyi, should we close this or is there still value in having a custom registrator even if we try to avoid needing it?

[jzhuge]

Ok to close.

[flyrain]

Are we going to replace guava maps with non-guava ones? There are bunches of place using guava's Immutable Map. For example, here are 4 guava maps in class TableMetadata. We hit this issue when we were trying to broadcast one of them.

    this.snapshotsById = indexAndValidateSnapshots(snapshots, lastSequenceNumber);
    this.schemasById = indexSchemas();
    this.specsById = indexSpecs(specs);
    this.sortOrdersById = indexSortOrders(sortOrders);

[jzhuge]

What version of kyro do you use? Newer versions support ImmutableMap.

[jzhuge]

https://github.com/magro/kryo-serializers/blob/master/src/main/java/de/javakaffee/kryoserializers/guava/ImmutableMapSerializer.java

[flyrain]

Thanks for the information. Nice to see you here, @jzhuge. We are using kryo-shaded-4.0.2.jar in Spark. Which version has the change in https://github.com/magro/kryo-serializers/blob/master/src/main/java/de/javakaffee/kryoserializers/guava/ImmutableMapSerializer.java?
Also, Spark 3.1.2 uses kryo v4 per this doc, https://spark.apache.org/docs/latest/tuning.html#data-serialization.

[jzhuge]

Oops, I referenced the wrong repo.

If possible, you can add the following code to KryoSerializer.newKryo():

    ImmutableListSerializer.registerSerializers(kryo);
    ImmutableMapSerializer.registerSerializers(kryo);

Be sure to add dependency on de.javakaffee:kryo-serializers.

[flyrain]

Hi @jzhuge, not sure I understand. Do I add these lines to my Spark application or Spark itself?

    ImmutableListSerializer.registerSerializers(kryo);
    ImmutableMapSerializer.registerSerializers(kryo);

[jzhuge]

Spark itself if you can