Я изо всех сил пытаюсь использовать пользовательские классы case для записи в Cassandra (2.1.6) с помощью Spark (1.4.0). До сих пор я пробовал это, используя DataStax spark-cassandra-connector 1.4.0-M1 и следующие классы case:
case class Event(event_id: String, event_name: String, event_url: String, time: Option[Long])
[...]
case class RsvpResponse(event: Event, group: Group, guests: Long, member: Member, mtime: Long, response: String, rsvp_id: Long, venue: Option[Venue])
Чтобы это работало, я также реализовал следующий преобразователь:
implicit object EventToUDTValueConverter extends TypeConverter[UDTValue] {
def targetTypeTag = typeTag[UDTValue]
def convertPF = {
case e: Event => UDTValue.fromMap(toMap(e)) // toMap just transforms the case class into a Map[String, Any]
}
}
TypeConverter.registerConverter(EventToUDTValueConverter)
Если я найду преобразователь вручную, я смогу использовать его для преобразования экземпляра Event в UDTValue, однако при использовании sc.saveToCassandra, передавая ему экземпляр RsvpResponse со связанными объектами, я получаю следующую ошибку:
15/06/23 23:56:29 ERROR Executor: Exception in task 1.0 in stage 0.0 (TID 1)
com.datastax.spark.connector.types.TypeConversionException: Cannot convert object Event(EVENT9136830076436652815,First event,http://www.meetup.com/first-event,Some(1435100185774)) of type class model.Event to com.datastax.spark.connector.UDTValue.
at com.datastax.spark.connector.types.TypeConverter$$anonfun$convert$1.apply(TypeConverter.scala:42)
at com.datastax.spark.connector.types.UserDefinedType$$anon$1$$anonfun$convertPF$1.applyOrElse(UserDefinedType.scala:33)
at com.datastax.spark.connector.types.TypeConverter$class.convert(TypeConverter.scala:40)
at com.datastax.spark.connector.types.UserDefinedType$$anon$1.convert(UserDefinedType.scala:31)
at com.datastax.spark.connector.writer.DefaultRowWriter$$anonfun$readColumnValues$2.apply(DefaultRowWriter.scala:46)
at com.datastax.spark.connector.writer.DefaultRowWriter$$anonfun$readColumnValues$2.apply(DefaultRowWriter.scala:43)
Кажется, мой конвертер никогда не вызывается из-за того, как библиотека коннекторов обрабатывает UDTValue внутри. Однако описанное выше решение работает для чтения данных из таблиц Cassandra (включая пользовательские типы). Основываясь на документах по коннектору, я также заменил мои вложенные классы case с типами com.datastax.spark.connector.UDTValue напрямую, которые затем устраняют описанную проблему, но прерывают чтение данных. Я не могу представить, что я должен определить две отдельные модели для чтения и записи данных. Или я пропустил что-то очевидное здесь?
