1

Estoy trabajando con Spark en Scala.

Mi objetivo es almacenar un RDD en un fichero CSV.

El RDD lo construí a partir de una clase que me creé:

case class SensorAverage(
    sensor: String, 
    date: String, 
    values: Array[Double], 
    average : Double 
) 

El array del que se compone es grande (debe tener unos 100 valores). Por ello descarté definir la estructura con la función schema.

Una vez tengo el RDD, lo transformo a un dataframe y lo intento escribir en un fichero, pero da error porque no puede interpretar bien la lista.

Adjunto parte del código:

var average = 0.0

val outRdd = sensorRdd.map( row =>
      {
        [...]
         SensorAverage(row.sensor, row.date, row.values, average)
      }
      
      )
      
outRdd.foreach(println)
val outdf = session.createDataFrame(outRdd)

outdf.write.mode(SaveMode.Overwrite).csv("./out.csv");

sensorRdd es otro RDD que tiene datos que siguen esta estructura:

case class SensorValue(
    sensor: String, 
    date: String, 
    values: Array[Double]
)

El error da en esta línea:

outdf.write.mode(SaveMode.Overwrite).csv("./out.csv");

El error concretamente es este:

Exception in thread "main" org.apache.spark.sql.AnalysisException: CSV data source does not support array data type.

Se me ha ocurrido que si tuviera un rdd o un dataframe que tuviera los elementos de la lista en columnas podría solucionar el problema. Pero como he dicho la lista es demasiado grande para hacer el schema.

¿Alguien sabría cómo construir el rdd/dataframe sin la lista y que todos sus elementos estuvieran en una columna?

Un saludo y muchas gracias

1 respuesta 1

0

El problema está en que alguno de los valores en un array, y el formato CSV no permite el tipo array, así que propongo que lo convirtamos a una cadena de texto y así lo almacenemos.

En general, digamos que tienes un RDD de la forma:

1,1,Seq(1,1)
2,2,Seq(2,2)

Osea, los campos 1 y 2 son Int y el tercero es un Array.

Primero importemos los implicits y las funciones:

import org.apache.spark.sql.functions._
import spark.implicits._

// Creamos nuestro RDD de prueba:

val myRdd = spark.sparkContext.parallelize(
    Seq(
        (1,1,Seq(1,1)),
        (2,2,Seq(2,2))
    )
)

// Lo convertimos a DataFrame

myRdd.toDF
    .withColumn("_3", col("_3").cast("string"))
    .write
    .csv("/tmp/spark-test/csv_test")

Nota como al menos tenemos que conocer la posición del array problemático, que en nuestro ejemplo es "_3".

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.