Comment imprimer le contenu de RDD?

J'essaie d'imprimer le contenu d'une collection sur la console Spark.

J'ai un type:

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

Et j'utilise la commande:

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

Mais ceci est imprimé:

res1: org.apache.spark.rdd.RDD [Unit] = MappedRDD [4] sur la carte à: 19

Comment puis-je écrire le RDD sur la console ou l'enregistrer sur le disque afin de pouvoir afficher son contenu?

Si vous souhaitez afficher le contenu d'un RDD, une méthode consiste à utiliser collect():

myRDD.collect().foreach(println)

Ce n'est pas une bonne idée, cependant, lorsque le RDD a des milliards de lignes. Utilisez take()pour n'en prendre que quelques-uns à imprimer:

myRDD.take(n).foreach(println)

La mapfonction est une transformation , ce qui signifie que Spark n'évaluera pas réellement votre RDD jusqu'à ce que vous exécutiez une action dessus.

Pour l'imprimer, vous pouvez utiliser foreach(qui est une action):

linesWithSessionId.foreach(println)

Pour l'écrire sur le disque, vous pouvez utiliser l'une des saveAs...fonctions (actions fixes) de l' API RDD

Si vous l'exécutez sur un cluster, il printlnne sera pas imprimé dans votre contexte. Vous devez apporter les RDDdonnées à votre session. Pour ce faire, vous pouvez le forcer sur la baie locale, puis l'imprimer:

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

Vous pouvez convertir votre RDDen un DataFramealors show()il.

// 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()

Cela affichera les 20 premières lignes de vos données, donc la taille de vos données ne devrait pas être un problème.

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

Il existe probablement de nombreuses différences architecturales entre myRDD.foreach(println)et myRDD.collect().foreach(println)(non seulement «collecter», mais aussi d'autres actions). L'une des différences que j'ai constatées est que lors de l'exécution myRDD.foreach(println), la sortie sera dans un ordre aléatoire. Par exemple: si mon rdd provient d'un fichier texte où chaque ligne a un numéro, la sortie aura un ordre différent. Mais quand je l'ai fait myRDD.collect().foreach(println), l'ordre reste exactement comme le fichier texte.

En python

   linesWithSessionIdCollect = linesWithSessionId.collect()
   linesWithSessionIdCollect

Cela imprimera tout le contenu du RDD

c.take(10)

et la version plus récente de Spark affichera bien la table.

Au lieu de taper à chaque fois, vous pouvez;

[1] Créez une méthode d'impression générique dans Spark Shell.

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

[2] Ou mieux encore, en utilisant des implicits, vous pouvez ajouter la fonction à la classe RDD pour imprimer son contenu.

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

Exemple d'utilisation:

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

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

Production:

2
6
4
8

Important

Cela n'a de sens que si vous travaillez en mode local et avec une petite quantité de données. Sinon, vous ne pourrez pas voir les résultats sur le client ou vous manquerez de mémoire en raison du résultat du grand ensemble de données.

Vous pouvez également enregistrer sous forme de fichier: rdd.saveAsTextFile("alicia.txt")

Dans la syntaxe java:

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