Как реализовать карту fastutils в UDAF Spark?

Question

Я строю UDAF Spark, где храню промежуточные данные в карте fastutils.Схема выглядит следующим образом:

def bufferSchema = new StructType().add("my_map_col", MapType(StringType, IntegerType))

Я инициализируюсь без проблем:

def initialize(buffer: MutableAggregationBuffer) = {
   buffer(0) = new Object2IntOpenHashMap[String]()
}

При попытке обновления возникает проблема:

def update(buffer: MutableAggregationBuffer, input: Row) = { 
  val myMap = buffer.getAs[Object2IntOpenHashMap[String]](0)
  myMap.put(input.getAs[String](0), 1)
  buffer(0) = myMap
}

Получение следующей ошибки:

Caused by: java.lang.ClassCastException: scala.collection.immutable.Map$EmptyMap$ cannot be cast to it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap

Как я могу сделать эту работу?

Answer 1

Как я могу заставить эту работу работать?

Не совсем.Этот

buffer.getAs[Object2IntOpenHashMap[String]](0)

эквивалентен

buffer.get(0).asInstanceOf[Object2IntOpenHashMap[String]]]

и внешний тип для MapType: scala.collection.Map.

На практике этов любом случае это тупик - UserDefinedAggregate функции делают полную копию данных по каждому вызову .Возможно, вам повезет больше с Aggregator (как в связанном вопросе).