Bagaimana cara mencetak konten RDD?

Saya mencoba mencetak konten koleksi ke konsol Spark.

Saya punya tipe:

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

Dan saya menggunakan perintah:

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

Tapi ini dicetak:

res1: org.apache.spark.rdd.RDD [Unit] = MappedRDD [4] di peta pada: 19

Bagaimana cara menulis RDD ke konsol atau menyimpannya ke disk sehingga saya dapat melihat isinya?

Jika Anda ingin melihat konten RDD, salah satu caranya adalah dengan menggunakan collect():

myRDD.collect().foreach(println)

Itu bukanlah ide yang baik, meskipun, RDD memiliki milyaran baris. Gunakan take()untuk mengambil sedikit saja untuk dicetak:

myRDD.take(n).foreach(println)

The mapfungsi adalah transformasi , yang berarti bahwa Spark tidak akan benar-benar mengevaluasi RDD Anda sampai Anda menjalankan aksi di atasnya.

Untuk mencetaknya, Anda dapat menggunakan foreach(yang merupakan tindakan):

linesWithSessionId.foreach(println)

Untuk menulisnya ke disk, Anda dapat menggunakan salah satu saveAs...fungsi (tindakan diam) dari RDD API

Jika Anda menjalankan ini di cluster maka printlntidak akan mencetak kembali ke konteks Anda. Anda perlu membawa RDDdata ke sesi Anda. Untuk melakukan ini, Anda dapat memaksanya ke array lokal dan kemudian mencetaknya:

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

Anda dapat mengubah Anda RDDmenjadi DataFramelalu show()itu.

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

Ini akan menampilkan 20 baris teratas dari data Anda, jadi ukuran data Anda seharusnya tidak menjadi masalah.

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

Mungkin ada banyak perbedaan arsitektural antara myRDD.foreach(println)dan myRDD.collect().foreach(println)(tidak hanya 'kumpulkan', tetapi juga tindakan lainnya). Salah satu perbedaan yang saya lihat adalah ketika melakukan myRDD.foreach(println), outputnya akan dalam urutan acak. Misalnya: jika rdd saya berasal dari file teks di mana setiap baris memiliki nomor, output akan memiliki urutan yang berbeda. Tetapi ketika saya melakukannya myRDD.collect().foreach(println), urutan tetap seperti file teks.

Dengan python

   linesWithSessionIdCollect = linesWithSessionId.collect()
   linesWithSessionIdCollect

Ini akan mencetak semua konten RDD

c.take(10)

dan Spark versi yang lebih baru akan menampilkan tabel dengan baik.

Alih-alih mengetik setiap kali, Anda bisa;

[1] Buat metode cetak umum di dalam Spark Shell.

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

[2] Atau bahkan lebih baik, menggunakan implikasinya, Anda dapat menambahkan fungsi ke kelas RDD untuk mencetak isinya.

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

Contoh penggunaan:

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

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

Keluaran:

2
6
4
8

Penting

Ini hanya masuk akal jika Anda bekerja dalam mode lokal dan dengan sedikit kumpulan data. Jika tidak, Anda tidak akan dapat melihat hasilnya di klien atau kehabisan memori karena hasil dataset yang besar.

Anda juga dapat menyimpan sebagai file: rdd.saveAsTextFile("alicia.txt")

Dalam sintaks java:

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