Início / coding / Perguntas / 79210262
Accepted
David Regan
Asked: 2024-11-21 16:25:28 +0800 CST

Problemas de conversão de estrutura Spark para classe case com getAs[T]

  • 772

Eu costumo usar a mapfunção em spark Dataset rows para fazer transformações em Scala em objetos tipados. Meu padrão usual é converter resultados intermediários criados a partir de transformações de dataframe ( withColumn, groupBy, etc.) e criar um Dataset tipado do resultado intermediário para que eu possa usar map.

Isso funciona bem, mas leva a muitas classes de casos "temporárias" para resultados intermediários ou tipos de tuplas difíceis de manejar.

Uma alternativa seria executar mapem um quadro de dados e recuperar campos digitados da linha usando, getAs[T]mas isso não parece funcionar com spark.implicitsif Té uma classe de caso.

Por exemplo, isso dá o erroClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to Person

import org.apache.spark._
import org.apache.spark.sql._
import org.apache.spark.sql.functions.{round => colRound, min => colMin, max => colMax, _}
import org.apache.spark.sql.expressions.Window

import spark.implicits._

final case class Person(name: String, age: Integer)
val people = Seq(Person("Alice", 20), Person("Bob", 30), Person("Charlie", 40)).toDS

val df = people.alias("p")
               .select($"p.name", struct($"p.*").alias("person"))
val ds = df.map(row => {
  val name = row.getAs[String]("name")
  val person = row.getAs[Person]("person")
  (name, person)
})

display(ds)

enquanto isso funciona bem:

import org.apache.spark._
import org.apache.spark.sql._
import org.apache.spark.sql.functions.{round => colRound, min => colMin, max => colMax, _}
import org.apache.spark.sql.expressions.Window

import spark.implicits._

final case class Person(name: String, age: Integer)
val people = Seq(Person("Alice", 20), Person("Bob", 30), Person("Charlie", 40)).toDS

val df = people.alias("p")
               .select($"p.name", struct($"p.*").alias("person"))
               .as[Tuple2[String, Person]]
val ds = df.map(row => {
  val name = row._1
  val person = row._2
  (name, person)
})

display(ds)

Então o spark está convertendo alegremente a estrutura person do dataframe para a Personclasse case no segundo exemplo, mas não fará isso no primeiro exemplo. Alguém sabe uma maneira simples de consertar isso?

Obrigado,

Davi

dataframe

1 respostas

  1. Best Answer

    "Simples", possivelmente :), mas usando muito api's internas que estão sujeitas a mudanças (e já mudaram). Este código não funcionará como está no Spark 4 também (testado no 3.5.1).

    Como uma abordagem, também é provável que seja mais lento do que o segundo exemplo que você fornece usando tuplas, pois o código Spark traduz de InternalRow para o usuário Land Row antes de inserir seu código de mapa. O código abaixo então converte de volta para InternalRow antes de chamar o decodificador.

    resolveAndBind normalmente funciona nesse tipo de exemplo, mas também não há garantia de que funcione em todos os casos, pois a resolução de nomes de campos, etc., normalmente precisa acontecer como parte da análise completa do plano de consulta.

    import org.apache.spark._
import org.apache.spark.sql._
import org.apache.spark.sql.functions.{round => colRound, min => colMin, max => colMax, _}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}

import spark.implicits._

implicit val pEnc = implicitly[Encoder[Person]].asInstanceOf[ExpressionEncoder[Person]]
val decoder = pEnc.resolveAndBind().objDeserializer

val people = Seq(Person("Alice", 20), Person("Bob", 30), Person("Charlie", 40)).toDS

val df = people.alias("p")
  .select($"p.name", struct($"p.*").alias("person"))
val ds = df.map(row => {
  val name = row.getAs[String]("name")
  val personRow = row.getAs[Row]("person")
  val person = decoder.eval(CatalystTypeConverters.convertToCatalyst(personRow).asInstanceOf[InternalRow]).asInstanceOf[Person]
  (name, person)
})

ds.show

    em resumo, é melhor usar um wrapper de tupla e a codificação interna sempre que possível, pois é mais rápido e foi projetado e testado para funcionar dessa maneira.

    • 0

relate perguntas