Требовать Kryo сериализации в Спарк (Scala)

Question

меня Kryo сериализации включен с этим:

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

Я хочу, чтобы гарантировать, что пользовательский класс сериализации с помощью Kryo когда перемешиваются между узлами. Я могу зарегистрировать класс с Kryo таким образом:

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

Как я понимаю, это на самом деле не гарантирует, что KYRO сериализации используется; если сериализатор недоступен, kryo вернется к сериализации Java.

Чтобы гарантировать, что Kryo Сериализация происходит, я последовал эту рекомендацию из документации Spark:

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

Но это приводит к IllegalArugmentException быть выброшено («Класс не зарегистрирован») для связки различных классов, которые я предположим, Спарк использует внутренне, например, следующее:

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

Конечно, я не должен вручную зарегистрировать каждый из этих отдельных классов с Kryo? Эти сериализаторы определены в kryo, так есть ли способ автоматически зарегистрировать их все?

Answer 1

Как я понимаю, на самом деле это не гарантирует, что используется сериализация кири; если сериализатор недоступен, kryo вернется к сериализации Java.

№ Если вы установили spark.serializer в org.apache.spark.serializer. KryoSerializer, то Spark будет использовать Kryo. Если Kryo недоступен, вы получите сообщение об ошибке. Нет возврата.

Итак, что же это за регистрация Крио?

Когда Kryo сериализует экземпляр незарегистрированного класса, он должен вывести полное имя класса. Это много персонажей. Вместо этого, если класс был предварительно зарегистрирован, Kryo может просто выводить числовую ссылку на этот класс, который составляет всего 1-2 байта.

Это особенно важно, когда каждый ряд RDD сериализуется с помощью Kryo. Вы не хотите включать одно и то же имя класса для каждого из миллиардов строк. Поэтому вы предварительно зарегистрируете эти классы. Но легко забыть зарегистрировать новый класс, а затем вы снова теряете байты. Решение состоит в том, чтобы требовать регистрации каждого класса:

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

Теперь Kryo никогда не будет выдавать полные имена классов. Если он встречает незарегистрированный класс, это ошибка времени выполнения.

К сожалению, сложно перечислить все классы, которые вы собираетесь сериализовать заранее. Идея заключается в том, что Spark регистрирует классы, специфичные для Spark, и вы регистрируете все остальное. У вас есть RDD[(X, Y, Z)]? Вам необходимо зарегистрироваться classOf[scala.Tuple3[_, _, _]].

В действительности list of classes that Spark registers содержит CompactBuffer, поэтому, если вы видите ошибку для этого, вы делаете что-то неправильно. Вы обходите процедуру регистрации Spark. Вы должны использовать либо spark.kryo.classesToRegister, либо spark.kryo.registrator, чтобы зарегистрировать свои классы. (См. config options. Если вы используете GraphX, регистратор должен позвонить по телефону GraphXUtils. registerKryoClasses.)

Answer 2

Основываясь на том, что вы видите, догадка, вы пропустили утверждение:

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

За последние несколько дней я также борюсь с преобразованием сериализации Kryo, в том числе для Graphx , включая регистрацию scala.Tuple3 с Kryo, по-видимому, потому что код Spark/GraphX ​​создает Tuple3, когда я делаю «sortBy».

Были добавлены кучу других классов, один за другим, чтобы зарегистрироваться для регистрации в Kryo, в основном Scala и Spark. Я бы не подумал, что мне нужно будет добавить. Думая/надеясь, что лучше использовать Kryo с Spark.