¿Cómo imprimir el contenido de RDD?

Estoy intentando imprimir el contenido de una colección en la consola Spark.

Tengo un tipo:

linesWithSessionId: org.apache.spark.rdd.RDD[String] = FilteredRDD[3]

Y uso el comando:

scala> linesWithSessionId.map(line => println(line))

Pero esto está impreso:

res1: org.apache.spark.rdd.RDD [Unidad] = MappedRDD [4] en el mapa en: 19

¿Cómo puedo escribir el RDD en la consola o guardarlo en el disco para poder ver su contenido?

Si desea ver el contenido de un RDD, una forma es usar collect():

myRDD.collect().foreach(println)

Sin embargo, esa no es una buena idea cuando el RDD tiene miles de millones de líneas. Use take()para tomar solo unos pocos para imprimir:

myRDD.take(n).foreach(println)

La mapfunción es una transformación , lo que significa que Spark en realidad no evaluará su RDD hasta que ejecute una acción sobre él.

Para imprimirlo, puede usar foreach(que es una acción):

linesWithSessionId.foreach(println)

Para escribirlo en el disco, puede usar una de las saveAs...funciones (aún acciones) de la API RDD

Si está ejecutando esto en un clúster println, no volverá a imprimir en su contexto. Necesita traer los RDDdatos a su sesión. Para hacer esto, puede forzarlo a una matriz local y luego imprimirlo:

linesWithSessionId.toArray().foreach(line => println(line))

Puede convertir su RDDa DataFrameluego show().

// For implicit conversion from RDD to DataFrame
import spark.implicits._

fruits = sc.parallelize([("apple", 1), ("banana", 2), ("orange", 17)])

// convert to DF then show it
fruits.toDF().show()

Esto mostrará las 20 líneas principales de sus datos, por lo que el tamaño de sus datos no debería ser un problema.

+------+---+                                                                    
|    _1| _2|
+------+---+
| apple|  1|
|banana|  2|
|orange| 17|
+------+---+

Probablemente hay muchas diferencias arquitectónicas entre myRDD.foreach(println)y myRDD.collect().foreach(println)(no solo 'recoger', sino también otras acciones). Una de las diferencias que vi es que al hacerlo myRDD.foreach(println), la salida será en orden aleatorio. Por ejemplo: si mi rdd proviene de un archivo de texto donde cada línea tiene un número, la salida tendrá un orden diferente. Pero cuando lo hice myRDD.collect().foreach(println), el orden permanece igual que el archivo de texto.

En pitón

   linesWithSessionIdCollect = linesWithSessionId.collect()
   linesWithSessionIdCollect

Esto imprimirá todo el contenido del RDD

c.take(10)

y la versión más nueva de Spark mostrará la tabla muy bien.

En lugar de escribir cada vez, puedes;

[1] Cree un método de impresión genérico dentro de Spark Shell.

def p(rdd: org.apache.spark.rdd.RDD[_]) = rdd.foreach(println)

[2] O incluso mejor, usando implicits, puedes agregar la función a la clase RDD para imprimir su contenido.

implicit class Printer(rdd: org.apache.spark.rdd.RDD[_]) {
    def print = rdd.foreach(println)
}

Ejemplo de uso:

val rdd = sc.parallelize(List(1,2,3,4)).map(_*2)

p(rdd) // 1
rdd.print // 2

Salida:

2
6
4
8

Importante

Esto solo tiene sentido si está trabajando en modo local y con una pequeña cantidad de conjunto de datos. De lo contrario, no podrá ver los resultados en el cliente o quedarse sin memoria debido al resultado del gran conjunto de datos.

También puede guardar como archivo: rdd.saveAsTextFile("alicia.txt")

En la sintaxis de Java:

rdd.collect().forEach(line -> System.out.println(line));