Проблема с равенством Spark DataType для встроенных типов Spark

Question

Во время работы приложения искры я получаю ошибки глубоко внутри катализатора.

Например:

java.lang.RuntimeException: scala.MatchError: LongType (of class org.apache.spark.sql.types.LongType$)
org.apache.spark.sql.catalyst.expressions.Cast.org$apache$spark$sql$catalyst$expressions$Cast$$nullSafeCastFunction(Cast.scala:637)
org.apache.spark.sql.catalyst.expressions.Cast.doGenCode(Cast.scala:625)
org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$genCode$2.apply(Expression.scala:107)
org.apache.spark.sql.catalyst.expressions.Expression$$anonfun$genCode$2.apply(Expression.scala:104)
scala.Option.getOrElse(Option.scala:121)
org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:104)

Я сузил это до следующего внутри плана искры:

Project [if (isnull(_rawTime#348L)) null else UDF(toTime(_rawTime#348L)) AS _time#438,

(обратите внимание, что я не могу контролировать нулевую схему, поскольку я получаю этот базовый кадр данных из коннектора spark hbase)

Где toTime - UDF, принимающий long и создающий TimeStamp.Кажется, что катализатор не может сопоставить LongType, хотя оператор match имеет:

 case LongType => castToLongCode(from, ctx)

Интересно то, что когда я запускаю это в первый раз, он работает нормально.На втором запуске это имеет эту проблему.

Обратите внимание, что это выполняется через Apache Livy, поэтому основной сеанс спарка должен быть одинаковым между выполнениями.

Я поместил следующий код в начале своей работы.

  logger.info("----------")
  logger.info(LongType + " " + System.identityHashCode(LongType))
  logger.info(DataTypes.LongType + " " + System.identityHashCode(DataTypes.LongType))
  logger.info("Equal " + (DataTypes.LongType == LongType))
  logger.info("----------")

И затем запустив его, я вижу:

first run:
----------
LongType 1044985410
LongType 1044985410
Equal true
----------
second run:
----------
LongType 355475697
LongType 1044985410
Equal false
----------

Вы можете увидеть на прогоне 2,Объектно-ориентированный вызов LongType отличается от идентификатора, который был запущен в первый раз.

Комментарий Спарка предлагает людям использовать синглтоны в DataTypes.Например, DataTypes.LongType, что имеет смысл, поскольку кажется, что они остаются прежними.Тем не менее, собственный код spark использует не-синглтон.

LongType определяется как

/**
 * @since 1.3.0
 */
@InterfaceStability.Stable
case object LongType extends LongType

, в то время как DataTypes.LongType равно

public static final DataType LongType = LongType$.MODULE$;

, которое относится к первому (объект case).Имеет смысл, что синглтон останется постоянным.На самом деле искровой код говорит Please use the singleton DataTypes.LongType . .. несмотря на тот факт, что загрузка внутреннего искрового кода этого не делает.Для меня это похоже на ошибку.

Кажется очень странным, что Scala-код в Spark нормально скомпилируется, а затем потерпит неудачу с этим внезапным изменением идентичности типов.

Итак, мои вопросы:

• Какая рекомендация по использованию DataType в Spark?Должен ли я использовать синглтоны или не синглетоны?
• Что может вызвать изменение этой личности подо мной?

Answer 1

Я решил проблему.

Практически все экземпляры DataType определены в Scala как:

 * @since 1.3.0
 */
@InterfaceStability.Stable
case object LongType extends LongType

Но ... во многих местах Spark использует код Java, который получает типы данных с помощью синглетонов:

 * Gets the LongType object.
 */
public static final DataType LongType = LongType$.MODULE$;

LongType$.MODULE$; - это способ вызова объекта case из java land.

Но я сериализовал DataType для Ливи, используя Kryo, и Kryo внутренне повторно инициализирует LongType$.MODULE$;. В Scala ссылка, которую вы получаете, когда получаете случай, объект привязан не к первому созданному экземпляру, а к последнему созданному экземпляру.

Итак, сроки:

• время 0: DataTypes.LongType имеет ссылку на 1, LongType имеет ссылку на 1 также. (где ref просто указывает на ссылку)
• время 1: Крио десериализуется и таким образом восстанавливает объект. Однако синглтон DataTypes.LongType указывает на первый экземпляр. то есть DataTypes.LongType имеет ref 1, LongType ref 2
• time> = 2: наступает хаос - DataTypes не пройдет проверку на равенство.

Решение состоит в том, чтобы не передавать Case-объекты Kryo таким способом. Возможно, по какой-то причине мы неправильно используем Kryo, или нам нужно использовать twitter / chill.