scala - create - pyspark rdd to dataframe

Convierta RDD a Dataframe en Spark/Scala

Question

El RDD se ha creado con el formato Array[Array[String]] y tiene los siguientes valores:

 Array[Array[String]] = Array(Array(4580056797, 0, 2015-07-29 10:38:42, 0, 1, 1), Array(4580056797, 0, 2015-07-29 10:38:42, 0, 1, 1), Array(4580056797, 0, 2015-07-29 10:38:42, 0, 1, 1), Array(4580057445, 0, 2015-07-29 10:40:37, 0, 1, 1), Array(4580057445, 0, 2015-07-29 10:40:37, 0, 1, 1))

Quiero crear un DataFrame con el esquema:

val schemaString = "callId oCallId callTime duration calltype swId"

Próximos pasos:

scala> val rowRDD = rdd.map(p => Array(p(0), p(1), p(2),p(3),p(4),p(5).trim))
rowRDD: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[14] at map at <console>:39
scala> val calDF = sqlContext.createDataFrame(rowRDD, schema)

Da el siguiente error:

consola: 45: error: valor del método sobrecargado createDataFrame con alternativas: (rdd: org.apache.spark.api.java.JavaRDD [ ], beanClass: Class [ ]) org.apache.spark.sql.DataFrame (rdd: org. apache.spark.rdd.RDD [ ], beanClass: Class [ ]) org.apache.spark.sql.DataFrame (rowRDD: org.apache.spark.api.java.JavaRDD [org.apache.spark.sql.Row] , esquema: org.apache.spark.sql.types.StructType) org.apache.spark.sql.DataFrame (rowRDD: org.apache.spark.rdd.RDD [org.apache.spark.sql.Row], esquema: org.apache.spark.sql.types.StructType) org.apache.spark.sql.DataFrame no se puede aplicar a (org.apache.spark.rdd.RDD [Array [String]],
org.apache.spark.sql.types.StructType) val calDF = sqlContext.createDataFrame (rowRDD, schema)

Answer 1

Primero debe convertir el Array en una Row y luego definir el esquema. Asumí que la mayoría de tus campos son Long

    val rdd: RDD[Array[String]] = ???
    val rows: RDD[Row] = rdd map {
      case Array(callId, oCallId, callTime, duration, swId) =>
        Row(callId.toLong, oCallId.toLong, callTime, duration.toLong, swId.toLong)
    }

    object schema {
      val callId = StructField("callId", LongType)
      val oCallId = StructField("oCallId", StringType)
      val callTime = StructField("callTime", StringType)
      val duration = StructField("duration", LongType)
      val swId = StructField("swId", LongType)

      val struct = StructType(Array(callId, oCallId, callTime, duration, swId))
    }

    sqlContext.createDataFrame(rows, schema.struct)

Answer 2

Solo pegue en una spark-shell :

val a = 
  Array(
    Array("4580056797", "0", "2015-07-29 10:38:42", "0", "1", "1"), 
    Array("4580056797", "0", "2015-07-29 10:38:42", "0", "1", "1"))

val rdd = sc.makeRDD(a)

case class X(callId: String, oCallId: String, 
  callTime: String, duration: String, calltype: String, swId: String)

Luego map() sobre el RDD para crear instancias de la clase de caso, y luego cree el DataFrame usando toDF() :

scala> val df = rdd.map { 
  case Array(s0, s1, s2, s3, s4, s5) => X(s0, s1, s2, s3, s4, s5) }.toDF()
df: org.apache.spark.sql.DataFrame = 
  [callId: string, oCallId: string, callTime: string, 
    duration: string, calltype: string, swId: string]

Esto infiere el esquema de la clase de caso.

Entonces puedes continuar con:

scala> df.printSchema()
root
 |-- callId: string (nullable = true)
 |-- oCallId: string (nullable = true)
 |-- callTime: string (nullable = true)
 |-- duration: string (nullable = true)
 |-- calltype: string (nullable = true)
 |-- swId: string (nullable = true)

scala> df.show()
+----------+-------+-------------------+--------+--------+----+
|    callId|oCallId|           callTime|duration|calltype|swId|
+----------+-------+-------------------+--------+--------+----+
|4580056797|      0|2015-07-29 10:38:42|       0|       1|   1|
|4580056797|      0|2015-07-29 10:38:42|       0|       1|   1|
+----------+-------+-------------------+--------+--------+----+

Si desea usar toDF() en un programa normal (no en el spark-shell ), asegúrese (citado desde aquí ):

• Para import sqlContext.implicits._ justo después de crear el SQLContext
• Defina la clase de caso fuera del método usando toDF()

Answer 3

Supongo que su schema es, como en la Guía de chispa , de la siguiente manera:

val schema =
  StructType(
    schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))

Si observa la firma del createDataFrame , aquí está el que acepta un StructType como segundo argumento (para Scala)

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

Crea un DataFrame desde un RDD que contiene Filas usando el esquema dado.

Entonces acepta como primer argumento un RDD[Row] . Lo que tienes en rowRDD es un RDD[Array[String]] por lo que hay una discrepancia.

¿Necesita un RDD[Array[String]] ?

De lo contrario, puede usar lo siguiente para crear su marco de datos:

val rowRDD = rdd.map(p => Row(p(0), p(1), p(2),p(3),p(4),p(5).trim))