Тип столбца выводится как двоичный с типизированным UDAF - PullRequest
Новая
       40

Тип столбца выводится как двоичный с типизированным UDAF

0 голосов
/ 13 февраля 2019

Я пытаюсь реализовать типизированный UDAF, который возвращает сложный тип.Каким-то образом Spark не может определить тип столбца результата и заставляет его binary помещать в него сериализованные данные.Вот минимальный пример, который воспроизводит проблему 

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{SparkSession, Encoder, Encoders}

case class Data(key: Int)

class NoopAgg[I] extends Aggregator[I, Map[String, Int], Map[String, Int]] {
    override def zero: Map[String, Int] = Map.empty[String, Int]

    override def reduce(b: Map[String, Int], a: I): Map[String, Int] = b

    override def merge(b1: Map[String, Int], b2: Map[String, Int]): Map[String, Int] = b1

    override def finish(reduction: Map[String, Int]): Map[String, Int] = reduction

    override def bufferEncoder: Encoder[Map[String, Int]] = Encoders.kryo[Map[String, Int]]

    override def outputEncoder: Encoder[Map[String, Int]] = Encoders.kryo[Map[String, Int]]
}

object Question {
  def main(args: Array[String]): Unit = {
      val spark = SparkSession.builder().master("local").getOrCreate()

      val sc = spark.sparkContext

      import spark.implicits._

      val ds = sc.parallelize((1 to 10).map(i => Data(i))).toDS()

      val noop = new NoopAgg[Data]().toColumn

      val result = ds.groupByKey(_.key).agg(noop.as("my_sum").as[Map[String, Int]])

      result.printSchema()
  }
}

Он печатает 

root
 |-- value: integer (nullable = false)
 |-- my_sum: binary (nullable = true)

1 Ответ

0 голосов
/ 13 февраля 2019

Здесь нет никакого вывода вообще.Вместо этого вы получаете более или менее то, что вы просите.В частности, ошибка здесь: 

override def outputEncoder: Encoder[Map[String, Int]] = Encoders.kryo[Map[String, Int]]

Encoders.kryo означает, что вы применяете сериализацию общего назначения и возвращаете двоичный двоичный объект.Вводящая в заблуждение часть - .as[Map[String, Int]] - вопреки тому, что можно ожидать, она не проверяется статически.Что еще хуже, он даже не был предварительно проверен планировщиком запросов, и исключение времени выполнения - throw, только когда result вычисляется. 

result.first

org.apache.spark.sql.AnalysisException: cannot resolve '`my_sum`' due to data type mismatch: cannot cast binary to map<string,int>;
  at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$3.applyOrElse(CheckAnalysis.scala:115)
...

Вы должны предоставить конкретные Encoder вместо этого либо явно : 

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder  

def outputEncoder: Encoder[Map[String, Int]] = ExpressionEncoder()

, либо неявно 

class NoopAgg[I](implicit val enc: Encoder[Map[String, Int]]) extends Aggregator[I, Map[String, Int], Map[String, Int]] {
  ...
  override def outputEncoder: Encoder[Map[String, Int]] = enc
}

В качестве побочного эффекта это сделает as[Map[String, Int]] устаревшим, как тип возвращаемого значенияAggregator уже известно.

...