Как наиболее эффективно преобразовать строку Scala DataFrame в класс case?

DataFrame - это просто псевдоним типа Dataset [Row]. Эти операции также называются «нетипизированными преобразованиями» в отличие от «типизированных преобразований», которые идут со строго типизированными наборами данных Scala / Java.

Преобразование из набора данных [строка] в набор данных [человек] очень просто в программе Spark.

val DFtoProcess = SQLContext.sql("SELECT * FROM peoples WHERE name='test'")

На этом этапе Spark преобразует ваши данные в DataFrame = Dataset [Row], коллекцию универсального объекта Row, поскольку ему неизвестен точный тип.

// Create an Encoders for Java class (In my eg. Person is a JAVA class)
// For scala case class you can pass Person without .class reference
val personEncoder = Encoders.bean(Person.class) 

val DStoProcess = DFtoProcess.as[Person](personEncoder)

Теперь Spark преобразует Dataset[Row] -> Dataset[Person] зависящий от типа объект JVM Scala / Java в соответствии с требованиями класса Person.

Пожалуйста, обратитесь к приведенной ниже ссылке, предоставленной databricks для получения дополнительной информации.

https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html

Насколько я знаю, вы не можете преобразовать строку в класс case, но иногда я выбирал прямой доступ к полям строки, например

map(row => myCaseClass(row.getLong(0), row.getString(1), row.getDouble(2))

Я считаю, что это проще, особенно если конструктору класса case нужны только некоторые поля из строки.

scala> import spark.implicits._    
scala> val df = Seq((1, "james"), (2, "tony")).toDF("id", "name")
df: org.apache.spark.sql.DataFrame = [id: int, name: string]

scala> case class Student(id: Int, name: String)
defined class Student

scala> df.as[Student].collectAsList
res6: java.util.List[Student] = [Student(1,james), Student(2,tony)]

Здесь spark в spark.implicits._ - это ваш SparkSession. Если вы находитесь внутри REPL, сеанс уже определен как spark, в противном случае вам необходимо изменить имя, чтобы оно соответствовало вашему SparkSession.

Конечно, вы можете сопоставить объект Row с классом case. Предположим, ваш SchemaType имеет много полей, и вы хотите сопоставить некоторые из них с вашим классом case. Если у вас нет нулевых полей, вы можете просто сделать:

case class MyClass(a: Long, b: String, c: Int, d: String, e: String)

dataframe.map {
  case Row(a: java.math.BigDecimal, 
    b: String, 
    c: Int, 
    _: String,
    _: java.sql.Date, 
    e: java.sql.Date,
    _: java.sql.Timestamp, 
    _: java.sql.Timestamp, 
    _: java.math.BigDecimal, 
    _: String) => MyClass(a = a.longValue(), b = b, c = c, d = d.toString, e = e.toString)
}

Этот подход не работает в случае нулевых значений, а также требует, чтобы вы явно определяли тип каждого отдельного поля. Если вам нужно обрабатывать нулевые значения, вам нужно либо отбросить все строки, содержащие нулевые значения, выполнив

dataframe.na.drop()

Это приведет к удалению записей, даже если пустые поля не те, которые используются в вашем сопоставлении с шаблоном для вашего класса case. Или, если вы хотите справиться с этим, вы можете превратить объект Row в список, а затем использовать шаблон параметров:

case class MyClass(a: Long, b: String, c: Option[Int], d: String, e: String)

dataframe.map(_.toSeq.toList match {
  case List(a: java.math.BigDecimal, 
    b: String, 
    c: Int, 
    _: String,
    _: java.sql.Date, 
    e: java.sql.Date,
    _: java.sql.Timestamp, 
    _: java.sql.Timestamp, 
    _: java.math.BigDecimal, 
    _: String) => MyClass(
      a = a.longValue(), b = b, c = Option(c), d = d.toString, e = e.toString)
}

Проверьте этот проект на github Sparkz (), который вскоре представит множество библиотек для упрощения API-интерфейсов Spark и DataFrame и сделает их более ориентированными на функциональное программирование.