Scala Ошибка для Hashtable [String, String]

Question

Я пишу небольшой UDF

  val transform = udf((x: Array[Byte]) => {

    val mapper = new ObjectMapper() with ScalaObjectMapper
    val stream: InputStream  = new ByteArrayInputStream(x);
    val obs = new ObjectInputStream(stream)

    val stock = mapper.readValue(obs, classOf[util.Hashtable[String, String]])
    stock
  })

Где я получаю ошибку

java.lang.UnsupportedOperationException: Schema for type java.util.Hashtable[String,String] is not supported
  at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$schemaFor$1.apply(ScalaReflection.scala:809)
  at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$schemaFor$1.apply(ScalaReflection.scala:740)
  at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
  at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:926)
  at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:49)
  at org.apache.spark.sql.catalyst.ScalaReflection$.schemaFor(ScalaReflection.scala:739)
  at org.apache.spark.sql.catalyst.ScalaReflection$.schemaFor(ScalaReflection.scala:736)
  at org.apache.spark.sql.functions$.udf(functions.scala:3898)
  ... 59 elided

Может кто-нибудь помочь понять, почему это происходит?

Answer 1

Ошибка, которую вы получаете, просто означает, что Spark не понимает java га sh таблиц. Мы можем воспроизвести вашу ошибку с помощью этого простого UDF.

val gen = udf(() => new java.util.Hashtable[String, String]())

Spark пытается создать DataType (вставить схему искры) из java.util.Hashtable, который он не знает, как делать. Spark понимает карты scala. Действительно, следующий код

val gen2 = udf(() => Map("a" -> "b"))
spark.range(1).select(gen2()).show()

дает

+--------+
|   UDF()|
+--------+
|[a -> b]|
+--------+

Чтобы исправить первую UDF, кстати, свою, вы можете преобразовать Hashtable в карту scala. Преобразование HashMap можно легко сделать с помощью JavaConverters. Я не знаю ни одного простого способа сделать это с Hashtable, но вы можете сделать это следующим образом:

import collection.JavaConverters._
val gen3 = udf(() => {
    val table = new java.util.Hashtable[String, String]()
    table.put("a", "b")
    Map(table.entrySet.asScala.toSeq.map(x => x.getKey -> x.getValue) :_*)
})