Jak wydrukować zawartość RDD?

Próbuję wydrukować zawartość kolekcji do konsoli Spark.

Mam typ:

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

I używam polecenia:

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

Ale to jest drukowane:

res1: org.apache.spark.rdd.RDD [Unit] = zmapowanyRDD [4] na mapie w: 19

Jak mogę napisać RDD na konsoli lub zapisać go na dysku, aby móc przeglądać jego zawartość?

Jeśli chcesz wyświetlić zawartość RDD, jednym ze sposobów jest użycie collect():

myRDD.collect().foreach(println)

Nie jest to jednak dobry pomysł, skoro RDD ma miliardy linii. Użyj, take()aby wydrukować tylko kilka:

myRDD.take(n).foreach(println)

Ta mapfunkcja jest transformacją , co oznacza, że ​​Spark nie oceni twojego RDD, dopóki nie wykonasz na nim akcji .

Aby go wydrukować, możesz użyć foreach(co jest akcją):

linesWithSessionId.foreach(println)

Aby zapisać go na dysku, możesz użyć jednej z saveAs...funkcji (nadal działań) z RDD API

Jeśli uruchamiasz to w klastrze println, nie drukujesz z powrotem do twojego kontekstu. Musisz przynieść RDDdane do swojej sesji. Aby to zrobić, możesz zmusić go do lokalnej tablicy, a następnie wydrukować:

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

Można konwertować RDDDo DataFramepotem show()ona.

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

Spowoduje to wyświetlenie 20 pierwszych wierszy danych, więc rozmiar danych nie powinien stanowić problemu.

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

Prawdopodobnie istnieje wiele różnic architektonicznych pomiędzy myRDD.foreach(println)i myRDD.collect().foreach(println)(nie tylko „zbieraj”, ale także inne działania). Jedną z różnic, które zauważyłem, jest to myRDD.foreach(println), że dane wyjściowe będą w losowej kolejności. Na przykład: jeśli mój rdd pochodzi z pliku tekstowego, w którym każda linia ma numer, dane wyjściowe będą miały inną kolejność. Ale kiedy to zrobiłem myRDD.collect().foreach(println), porządek pozostaje taki, jak plik tekstowy.

W Pythonie

   linesWithSessionIdCollect = linesWithSessionId.collect()
   linesWithSessionIdCollect

Spowoduje to wydrukowanie całej zawartości RDD

c.take(10)

a nowsza wersja Spark ładnie pokaże tabelę.

Zamiast wpisywać za każdym razem, możesz;

[1] Utwórz ogólną metodę drukowania wewnątrz Spark Shell.

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

[2] Lub nawet lepiej, używając implicits, możesz dodać funkcję do klasy RDD, aby wydrukować jej zawartość.

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

Przykładowe użycie:

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

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

Wynik:

2
6
4
8

Ważny

Ma to sens tylko wtedy, gdy pracujesz w trybie lokalnym i z małą ilością zestawu danych. W przeciwnym razie albo nie będzie można zobaczyć wyników na kliencie, albo zabraknie pamięci z powodu wyniku dużego zbioru danych.

Możesz również zapisać jako plik: rdd.saveAsTextFile("alicia.txt")

W składni java:

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