Cómo convertir un objeto rdd a un marco de datos en spark

139

¿Cómo puedo convertir un RDD ( org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]) a un marco de datos org.apache.spark.sql.DataFrame? Convertí un marco de datos a rdd usando .rdd. Después de procesarlo, lo quiero de vuelta en el marco de datos. Cómo puedo hacer esto ?

usuario568109
fuente
manera de lograr esto en Spark 2.x
mrsrinivas

Respuestas:

88

SqlContexttiene una serie de createDataFramemétodos que crean un DataFramedado RDD. Me imagino que uno de estos funcionará para su contexto.

Por ejemplo:

def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame

Crea un DataFrame a partir de un RDD que contiene filas utilizando el esquema dado.

El arquetipo de Pablo
fuente
93

Este código funciona perfectamente desde Spark 2.x con Scala 2.11

Importar clases necesarias

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}

Crear SparkSessionobjeto, y aquí estáspark

val spark: SparkSession = SparkSession.builder.master("local").getOrCreate
val sc = spark.sparkContext // Just used to create test RDDs

Vamos RDDa hacerloDataFrame

val rdd = sc.parallelize(
  Seq(
    ("first", Array(2.0, 1.0, 2.1, 5.4)),
    ("test", Array(1.5, 0.5, 0.9, 3.7)),
    ("choose", Array(8.0, 2.9, 9.1, 2.5))
  )
)

Método 1

Utilizando SparkSession.createDataFrame(RDD obj).

val dfWithoutSchema = spark.createDataFrame(rdd)

dfWithoutSchema.show()
+------+--------------------+
|    _1|                  _2|
+------+--------------------+
| first|[2.0, 1.0, 2.1, 5.4]|
|  test|[1.5, 0.5, 0.9, 3.7]|
|choose|[8.0, 2.9, 9.1, 2.5]|
+------+--------------------+

Método 2

Usar SparkSession.createDataFrame(RDD obj)y especificar nombres de columna.

val dfWithSchema = spark.createDataFrame(rdd).toDF("id", "vals")

dfWithSchema.show()
+------+--------------------+
|    id|                vals|
+------+--------------------+
| first|[2.0, 1.0, 2.1, 5.4]|
|  test|[1.5, 0.5, 0.9, 3.7]|
|choose|[8.0, 2.9, 9.1, 2.5]|
+------+--------------------+

Método 3 (respuesta real a la pregunta)

De esta manera, la entrada rdddebe ser de tipo RDD[Row].

val rowsRdd: RDD[Row] = sc.parallelize(
  Seq(
    Row("first", 2.0, 7.0),
    Row("second", 3.5, 2.5),
    Row("third", 7.0, 5.9)
  )
)

crear el esquema

val schema = new StructType()
  .add(StructField("id", StringType, true))
  .add(StructField("val1", DoubleType, true))
  .add(StructField("val2", DoubleType, true))

Ahora aplique ambos rowsRddy schemaacreateDataFrame()

val df = spark.createDataFrame(rowsRdd, schema)

df.show()
+------+----+----+
|    id|val1|val2|
+------+----+----+
| first| 2.0| 7.0|
|second| 3.5| 2.5|
| third| 7.0| 5.9|
+------+----+----+
mrsrinivas
fuente
2
Gracias por mostrar las diferentes formas de usar createDataFrame de una manera comprensible
vatsug
el tercer método es útil en los bloques de datos, ya que otros no funcionan y dan un error
Narendra Maru
67

Suponiendo que su RDD [fila] se llama rdd, puede usar:

val sqlContext = new SQLContext(sc) 
import sqlContext.implicits._
rdd.toDF()
dtjones
fuente
26
Creo que no funciona para RDD [Row]. ¿Me estoy perdiendo algo?
Daniel de Paula
44
Desde Spark 2.0 SQLContext se reemplaza por SparkSession, pero la clase se mantiene en la base del código para compatibilidad con versiones anteriores (scaladoc). Usarlo arroja una advertencia de desaprobación.
tomaskazemekas
18

Nota: esta respuesta se publicó originalmente aquí

Estoy publicando esta respuesta porque me gustaría compartir detalles adicionales sobre las opciones disponibles que no encontré en las otras respuestas


Para crear un DataFrame a partir de un RDD de filas, hay dos opciones principales:

1) Como ya se señaló, puede utilizar el toDF()que puede importar import sqlContext.implicits._. Sin embargo, este enfoque solo funciona para los siguientes tipos de RDD:

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

(fuente: Scaladoc del SQLContext.implicitsobjeto)

La última firma en realidad significa que puede funcionar para un RDD de tuplas o un RDD de clases de casos (porque las tuplas y las clases de casos son subclases de scala.Product).

Entonces, para usar este enfoque para un RDD[Row], debes mapearlo a un RDD[T <: scala.Product]. Esto se puede hacer asignando cada fila a una clase de caso personalizada o a una tupla, como en los siguientes fragmentos de código:

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

o

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

El principal inconveniente de este enfoque (en mi opinión) es que debe establecer explícitamente el esquema del DataFrame resultante en la función de mapa, columna por columna. Tal vez esto se pueda hacer programáticamente si no conoce el esquema de antemano, pero las cosas pueden ponerse un poco confusas allí. Entonces, alternativamente, hay otra opción:


2) Puede usar createDataFrame(rowRDD: RDD[Row], schema: StructType)como en la respuesta aceptada, que está disponible en el objeto SQLContext . Ejemplo para convertir un RDD de un antiguo DataFrame:

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

Tenga en cuenta que no es necesario establecer explícitamente ninguna columna de esquema. Reutilizamos el antiguo esquema del DF, que es de StructTypeclase y puede ampliarse fácilmente. Sin embargo, este enfoque a veces no es posible, y en algunos casos puede ser menos eficiente que el primero.

Daniel de Paula
fuente
Gracias por el detalleimport sqlContext.implicits.
javadba
En el futuro, no publique respuestas idénticas a varias preguntas. Si las preguntas son duplicadas, publique una buena respuesta, luego vote o marque para cerrar la otra pregunta como duplicado. Si la pregunta no es un duplicado, adapte sus respuestas a la pregunta. Vea ¿Cómo escribo una buena respuesta? .
15

Suponga que tiene un DataFramey desea realizar alguna modificación en los datos de los campos convirtiéndolos en RDD[Row].

val aRdd = aDF.map(x=>Row(x.getAs[Long]("id"),x.getAs[List[String]]("role").head))

Para volver a convertir DataFramedesde RDD, necesitamos definir el tipo de estructura de RDD.

Si el tipo de datos era Long entonces, se volverá como LongTypeen la estructura.

Si Stringentonces StringTypeen estructura.

val aStruct = new StructType(Array(StructField("id",LongType,nullable = true),StructField("role",StringType,nullable = true)))

Ahora puede convertir el RDD a DataFrame utilizando el método createDataFrame .

val aNamedDF = sqlContext.createDataFrame(aRdd,aStruct)
Ajay Gupta
fuente
7

Aquí hay un ejemplo simple de convertir su Lista en Spark RDD y luego convertir ese Spark RDD en Dataframe.

Tenga en cuenta que he usado la REPL scala de Spark-shell para ejecutar el siguiente código, Aquí sc es una instancia de SparkContext que está implícitamente disponible en Spark-shell. Espero que responda tu pregunta.

scala> val numList = List(1,2,3,4,5)
numList: List[Int] = List(1, 2, 3, 4, 5)

scala> val numRDD = sc.parallelize(numList)
numRDD: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[80] at parallelize at <console>:28

scala> val numDF = numRDD.toDF
numDF: org.apache.spark.sql.DataFrame = [_1: int]

scala> numDF.show
+---+
| _1|
+---+
|  1|
|  2|
|  3|
|  4|
|  5|
+---+
Rashmit Rathod
fuente
Un dato curioso: esto deja de funcionar cuando su Lista es Doble, en lugar de int (o Long, String, <: Product).
Rick Moritz
No responde el OP: que habla de RDD [Fila]
javadba
6

Método 1: (Scala)

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._
val df_2 = sc.parallelize(Seq((1L, 3.0, "a"), (2L, -1.0, "b"), (3L, 0.0, "c"))).toDF("x", "y", "z")

Método 2: (Scala)

case class temp(val1: String,val3 : Double) 

val rdd = sc.parallelize(Seq(
  Row("foo",  0.5), Row("bar",  0.0)
))
val rows = rdd.map({case Row(val1:String,val3:Double) => temp(val1,val3)}).toDF()
rows.show()

Método 1: (Python)

from pyspark.sql import Row
l = [('Alice',2)]
Person = Row('name','age')
rdd = sc.parallelize(l)
person = rdd.map(lambda r:Person(*r))
df2 = sqlContext.createDataFrame(person)
df2.show()

Método 2: (Python)

from pyspark.sql.types import * 
l = [('Alice',2)]
rdd = sc.parallelize(l)
schema =  StructType([StructField ("name" , StringType(), True) , 
StructField("age" , IntegerType(), True)]) 
df3 = sqlContext.createDataFrame(rdd, schema) 
df3.show()

Extrajo el valor del objeto de fila y luego aplicó la clase de caso para convertir rdd a DF

val temp1 = attrib1.map{case Row ( key: Int ) => s"$key" }
val temp2 = attrib2.map{case Row ( key: Int) => s"$key" }

case class RLT (id: String, attrib_1 : String, attrib_2 : String)
import hiveContext.implicits._

val df = result.map{ s => RLT(s(0),s(1),s(2)) }.toDF
Aravind Krishnakumar
fuente
4

En versiones más nuevas de spark (2.0+)

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql._
import org.apache.spark.sql.types._

val spark = SparkSession
  .builder()
  .getOrCreate()
import spark.implicits._

val dfSchema = Seq("col1", "col2", "col3")
rdd.toDF(dfSchema: _*)
ozzieisaacs
fuente
1
sparkSession es solo un contenedor para sqlContext, hiveContext
Archit
1
One needs to create a schema, and attach it to the Rdd.

Suponiendo que val spark es un producto de un SparkSession.builder ...

    import org.apache.spark._
    import org.apache.spark.sql._       
    import org.apache.spark.sql.types._

    /* Lets gin up some sample data:
     * As RDD's and dataframes can have columns of differing types, lets make our
     * sample data a three wide, two tall, rectangle of mixed types.
     * A column of Strings, a column of Longs, and a column of Doubules 
     */
    val arrayOfArrayOfAnys = Array.ofDim[Any](2,3)
    arrayOfArrayOfAnys(0)(0)="aString"
    arrayOfArrayOfAnys(0)(1)=0L
    arrayOfArrayOfAnys(0)(2)=3.14159
    arrayOfArrayOfAnys(1)(0)="bString"
    arrayOfArrayOfAnys(1)(1)=9876543210L
    arrayOfArrayOfAnys(1)(2)=2.71828

    /* The way to convert an anything which looks rectangular, 
     * (Array[Array[String]] or Array[Array[Any]] or Array[Row], ... ) into an RDD is to 
     * throw it into sparkContext.parallelize.
     * http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkContext shows
     * the parallelize definition as 
     *     def parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism)
     * so in our case our ArrayOfArrayOfAnys is treated as a sequence of ArraysOfAnys.
     * Will leave the numSlices as the defaultParallelism, as I have no particular cause to change it. 
     */
    val rddOfArrayOfArrayOfAnys=spark.sparkContext.parallelize(arrayOfArrayOfAnys)

    /* We'll be using the sqlContext.createDataFrame to add a schema our RDD.
     * The RDD which goes into createDataFrame is an RDD[Row] which is not what we happen to have.
     * To convert anything one tall and several wide into a Row, one can use Row.fromSeq(thatThing.toSeq)
     * As we have an RDD[somethingWeDontWant], we can map each of the RDD rows into the desired Row type. 
     */     
    val rddOfRows=rddOfArrayOfArrayOfAnys.map(f=>
        Row.fromSeq(f.toSeq)
    )

    /* Now to construct our schema. This needs to be a StructType of 1 StructField per column in our dataframe.
     * https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.types.StructField shows the definition as
     *   case class StructField(name: String, dataType: DataType, nullable: Boolean = true, metadata: Metadata = Metadata.empty)
     * Will leave the two default values in place for each of the columns:
     *        nullability as true, 
     *        metadata as an empty Map[String,Any]
     *   
     */

    val schema = StructType(
        StructField("colOfStrings", StringType) ::
        StructField("colOfLongs"  , LongType  ) ::
        StructField("colOfDoubles", DoubleType) ::
        Nil
    )

    val df=spark.sqlContext.createDataFrame(rddOfRows,schema)
    /*
     *      +------------+----------+------------+
     *      |colOfStrings|colOfLongs|colOfDoubles|
     *      +------------+----------+------------+
     *      |     aString|         0|     3.14159|
     *      |     bString|9876543210|     2.71828|
     *      +------------+----------+------------+
    */ 
    df.show 

Los mismos pasos, pero con menos declaraciones val:

    val arrayOfArrayOfAnys=Array(
        Array("aString",0L         ,3.14159),
        Array("bString",9876543210L,2.71828)
    )

    val rddOfRows=spark.sparkContext.parallelize(arrayOfArrayOfAnys).map(f=>Row.fromSeq(f.toSeq))

    /* If one knows the datatypes, for instance from JDBC queries as to RDBC column metadata:
     * Consider constructing the schema from an Array[StructField].  This would allow looping over 
     * the columns, with a match statement applying the appropriate sql datatypes as the second
     *  StructField arguments.   
     */
    val sf=new Array[StructField](3)
    sf(0)=StructField("colOfStrings",StringType)
    sf(1)=StructField("colOfLongs"  ,LongType  )
    sf(2)=StructField("colOfDoubles",DoubleType)        
    val df=spark.sqlContext.createDataFrame(rddOfRows,StructType(sf.toList))
    df.show
teserecter
fuente
1

Traté de explicar la solución usando el problema de conteo de palabras . 1. Lea el archivo usando sc

  1. Producir recuento de palabras
  2. Métodos para crear DF

    • Método rdd.toDF
    • rdd.toDF ("palabra", "cuenta")
      • spark.createDataFrame (rdd, esquema)

    Leer archivo usando chispa

    val rdd=sc.textFile("D://cca175/data/")  

    Rdd a Dataframe

    val df = sc.textFile ("D: // cca175 / data /") .toDF ("t1") df.show

    Método 1

    Crear el recuento de palabras RDD a Dataframe

    val df=rdd.flatMap(x=>x.split(" ")).map(x=>(x,1)).reduceByKey((x,y)=>(x+y)).toDF("word","count")

    Método 2

    Crear marco de datos desde Rdd

    val df=spark.createDataFrame(wordRdd) 
    # with header   
    val df=spark.createDataFrame(wordRdd).toDF("word","count")  df.show

    Método 3

    Definir esquema

    import org.apache.spark.sql.types._

    esquema de val = nuevo StructType (). add (StructField ("palabra", StringType, true)). add (StructField ("cuenta", StringType, verdadero))

    Crear RowRDD

    import org.apache.spark.sql.Row
    val rowRdd=wordRdd.map(x=>(Row(x._1,x._2)))     

    Crear DataFrame desde RDD con esquema

    val df = spark.createDataFrame (rowRdd, esquema)
    df.show

Priyanshu Singh
fuente
0

Para convertir un Array [Row] a DataFrame o Dataset, lo siguiente funciona con elegancia:

Digamos que el esquema es el StructType para la fila, luego

val rows: Array[Row]=...
implicit val encoder = RowEncoder.apply(schema)
import spark.implicits._
rows.toDS
Tom
fuente