Как распечатать содержимое RDD?

Я пытаюсь распечатать содержимое коллекции на консоли Spark.

У меня тип:

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

И я использую команду:

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

Но это напечатано:

res1: org.apache.spark.rdd.RDD [Unit] = MappedRDD [4] на карте по адресу: 19

Как мне записать RDD на консоль или сохранить его на диск, чтобы я мог просматривать его содержимое?

Если вы хотите просмотреть содержимое RDD, можно использовать collect():

myRDD.collect().foreach(println)

Однако это плохая идея, когда в RDD миллиарды строк. Используйте, take()чтобы распечатать всего несколько:

myRDD.take(n).foreach(println)

mapФункция является преобразованием , а это значит , что искра не будет на самом деле оценить вашу RDD , пока вы не запустите действие на него.

Чтобы распечатать его, вы можете использовать foreach(что является действием):

linesWithSessionId.foreach(println)

Чтобы записать его на диск, вы можете использовать одну из saveAs...функций (статических действий) из RDD API.

Если вы запускаете это в кластере println, печать обратно в ваш контекст не выполняется . Вам необходимо перенести RDDданные в вашу сессию. Для этого вы можете принудительно поместить его в локальный массив, а затем распечатать:

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

Вы можете преобразовать ваш RDDв a DataFramethen show()it.

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

Это покажет верхние 20 строк ваших данных, поэтому размер ваших данных не должен быть проблемой.

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

Вероятно, существует множество архитектурных различий между myRDD.foreach(println)и myRDD.collect().foreach(println)(не только «сбор», но и другие действия). Одно из различий, которое я заметил, заключается в том, что при выполнении myRDD.foreach(println)вывод будет в случайном порядке. Например: если мой rdd поступает из текстового файла, в котором каждая строка имеет номер, вывод будет иметь другой порядок. Но когда я это сделал myRDD.collect().foreach(println), порядок остается таким же, как и в текстовом файле.

В питоне

   linesWithSessionIdCollect = linesWithSessionId.collect()
   linesWithSessionIdCollect

Будет распечатано все содержимое СДР.

c.take(10)

и более новая версия Spark будет хорошо отображать таблицу.

Вместо того, чтобы печатать каждый раз, вы можете;

[1] Создайте общий метод печати внутри Spark Shell.

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

[2] Или, что еще лучше, используя неявные выражения, вы можете добавить функцию в класс RDD, чтобы распечатать его содержимое.

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

Пример использования:

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

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

Вывод:

2
6
4
8

Важный

Это имеет смысл только в том случае, если вы работаете в локальном режиме и с небольшим набором данных. В противном случае вы либо не сможете увидеть результаты на клиенте, либо закончится память из-за большого набора данных.

Вы также можете сохранить как файл: rdd.saveAsTextFile("alicia.txt")

В синтаксисе java:

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