Как преобразовать RDD [Row] в DataFrame

Я играл с конвертированием RDD в DataFrames и обратно. Во-первых, у меня был RDD типа (Int, Int), называемый dataPair. Затем я создал объект DataFrame с заголовками столбцов, используя:

val dataFrame = dataPair.toDF(header(0), header(1))

Затем я преобразовал его из DataFrame обратно в RDD, используя:

val testRDD = dataFrame.rdd

который возвращает RDD типа org.apache.spark.sql.Row(not (Int, Int)). Затем я хотел бы преобразовать его обратно в RDD с использованием .toDF, но я получаю сообщение об ошибке:

error: value toDF is not a member of org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]

Я попытался определить схему типа Data (Int, Int) для testRDD, но я получаю исключения типа несоответствия:

error: type mismatch;
found   : org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
required: org.apache.spark.rdd.RDD[Data]
    val testRDD: RDD[Data] = dataFrame.rdd
                                       ^

Я уже импортировал

import sqlContext.implicits._

Ответ 1

Чтобы создать DataFrame из RDD строк, обычно у вас есть два основных варианта:

1) Вы можете использовать toDF(), который можно импортировать с помощью import sqlContext.implicits._. Однако этот подход работает только для следующих типов RDD:

RDD[Int]
RDD[Long]
RDD[String]
RDD[T <: scala.Product]

(источник: Scaladoc объекта SQLContext.implicits)

Последняя сигнатура на самом деле означает, что она может работать для RDD кортежей или RDD классов case (поскольку кортежи и классы case являются подклассами scala.Product).

Итак, чтобы использовать этот подход для RDD[Row], вам нужно сопоставить его с RDD[T <: scala.Product]. Это можно сделать, сопоставляя каждую строку с пользовательским классом case или с кортежем, как в следующих фрагментах кода:

val df = rdd.map({ 
  case Row(val1: String, ..., valN: Long) => (val1, ..., valN)
}).toDF("col1_name", ..., "colN_name")

или

case class MyClass(val1: String, ..., valN: Long = 0L)
val df = rdd.map({ 
  case Row(val1: String, ..., valN: Long) => MyClass(val1, ..., valN)
}).toDF("col1_name", ..., "colN_name")

Основным недостатком этого подхода (на мой взгляд) является то, что вы должны явно установить схему результирующего DataFrame в функции map, по столбцу. Возможно, это можно сделать программно, если вы заранее не знаете схему, но там может быть немного грязно. Итак, альтернативно, есть еще один вариант:

2) Вы можете использовать createDataFrame(rowRDD: RDD[Row], schema: StructType), который доступен в SQLContext. Пример:

val df = oldDF.sqlContext.createDataFrame(rdd, oldDF.schema)

Обратите внимание, что нет необходимости явно устанавливать какой-либо столбец схемы. Мы повторно используем старую схему DF, которая имеет класс StructType и может быть легко расширена. Однако этот подход иногда невозможен, и в некоторых случаях он может быть менее эффективным, чем первый.

Я надеюсь, что это станет яснее, чем раньше. Приветствия.