scala - 尝试将数据帧行映射到更新行时出现编码器错误

Question

当我试图在我的代码中做同样的事情时，如下所述

dataframe.map(row => {
  val row1 = row.getAs[String](1)
  val make = if (row1.toLowerCase == "tesla") "S" else row1
  Row(row(0),make,row(2))
})

我从这里获取了上述引用:
Scala: How can I replace value in Dataframs using scala
但我收到编码器错误

Unable to find encoder for type stored in a Dataset. Primitive types (Int, S tring, etc) and Product types (case classes) are supported by importing spark.im plicits._ Support for serializing other types will be added in future releases.

注意:我使用的是 spark 2.0!

Answer 1

这里没有什么意外的。您正在尝试使用使用 Spark 1.x 编写且在 Spark 2.0 中不再受支持的代码:

在 1.x DataFrame.map是 ((Row) ⇒ T)(ClassTag[T]) ⇒ RDD[T]

在 2.x Dataset[Row].map是 ((Row) ⇒ T)(Encoder[T]) ⇒ Dataset[T]

老实说，它在 1.x 中也没有多大意义。独立于版本，您可以简单地使用 DataFrame接口(interface):

import org.apache.spark.sql.functions.{when, lower}

val df = Seq(
  (2012, "Tesla", "S"), (1997, "Ford", "E350"),
  (2015, "Chevy", "Volt")
).toDF("year", "make", "model")

df.withColumn("make", when(lower($"make") === "tesla", "S").otherwise($"make"))

如果你真的想用 map你应该使用静态类型的 Dataset :

import spark.implicits._

case class Record(year: Int, make: String, model: String)

df.as[Record].map {
  case tesla if tesla.make.toLowerCase == "tesla" => tesla.copy(make = "S")
  case rec => rec
}

或至少返回一个具有隐式编码器的对象:

df.map {
  case Row(year: Int, make: String, model: String) => 
    (year, if(make.toLowerCase == "tesla") "S" else make, model)
}

最后如果对于一些 完全疯了你真的想映射到 Dataset[Row] 的原因您必须提供所需的编码器:

import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row

// Yup, it would be possible to reuse df.schema here
val schema = StructType(Seq(
  StructField("year", IntegerType),
  StructField("make", StringType),
  StructField("model", StringType)
))

val encoder = RowEncoder(schema)

df.map {
  case Row(year, make: String, model) if make.toLowerCase == "tesla" => 
    Row(year, "S", model)
  case row => row
} (encoder)