Encoder error while trying to map dataframe row to updated row

by

There is nothing unexpected here. You’re trying to use code which has been written with Spark 1.x and is no longer supported in Spark 2.0:

  • in 1.x DataFrame.map is ((Row) ⇒ T)(ClassTag[T]) ⇒ RDD[T]
  • in 2.x Dataset[Row].map is ((Row) ⇒ T)(Encoder[T]) ⇒ Dataset[T]

To be honest it didn’t make much sense in 1.x either. Independent of version you can simply use DataFrame API:

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"))

If you really want to use map you should use statically typed 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
}

or at least return an object which will have implicit encoder:

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

Finally if for some completely crazy reason you really want to map over Dataset[Row] you have to provide required encoder:

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)

More Related Contents:

  1. How to create a Dataset of Maps?
  2. How to store custom objects in Dataset?
  3. Spark 2.0 Dataset vs DataFrame
  4. Why is “Unable to find encoder for type stored in a Dataset” when creating a dataset of custom case class?
  5. Perform a typed join in Scala with Spark Datasets
  6. Encode an ADT / sealed trait hierarchy into Spark DataSet column
  7. How to get keys and values from MapType column in SparkSQL DataFrame
  8. Spark – load CSV file as DataFrame?
  9. How to define partitioning of DataFrame?
  10. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  11. How can I change column types in Spark SQL’s DataFrame?
  12. SparkSQL: How to deal with null values in user defined function?
  13. How to aggregate values into collection after groupBy?
  14. Dropping a nested column from Spark DataFrame
  15. How to read records in JSON format from Kafka using Structured Streaming?
  16. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  17. Spark DataFrame: does groupBy after orderBy maintain that order?
  18. Filling gaps in timeseries Spark
  19. Spark UDAF with ArrayType as bufferSchema performance issues
  20. Apache Spark how to append new column from list/array to Spark dataframe
  21. How to use COGROUP for large datasets
  22. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  23. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  24. Create new Dataframe with empty/null field values
  25. DataFrame-ified zipWithIndex
  26. Spark: Add column to dataframe conditionally
  27. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  28. Calculate Cosine Similarity Spark Dataframe
  29. Why is join not possible after show operator?
  30. How to create DataFrame from Scala’s List of Iterables?

Leave a Comment