Spark - partisi ulang () vs coalesce ()

254

Menurut Learning Spark

Ingatlah bahwa mempartisi ulang data Anda adalah operasi yang cukup mahal. Spark juga memiliki versi yang repartition()dipanggil coalesce()yang dioptimalkan yang memungkinkan menghindari perpindahan data, tetapi hanya jika Anda mengurangi jumlah partisi RDD.

Satu perbedaan yang saya dapatkan adalah bahwa dengan repartition()jumlah partisi dapat ditingkatkan / dikurangi, tetapi dengan coalesce()jumlah partisi hanya dapat dikurangi.

Jika partisi tersebar di beberapa mesin dan coalesce()dijalankan, bagaimana bisa menghindari pergerakan data?

apache-spark distributed-computing rdd Praveen Sripati
sumber

Jawaban:

354

Ini menghindari shuffle penuh . Jika diketahui bahwa jumlahnya menurun maka pelaksana dapat dengan aman menyimpan data pada jumlah minimum partisi, hanya memindahkan data dari node tambahan, ke node yang kami simpan.

Jadi, akan seperti ini:

Node 1 = 1,2,3
Node 2 = 4,5,6
Node 3 = 7,8,9
Node 4 = 10,11,12

Lalu coalesceke 2 partisi:

Node 1 = 1,2,3 + (10,11,12)
Node 3 = 7,8,9 + (4,5,6)

Perhatikan bahwa Node 1 dan Node 3 tidak memerlukan data aslinya untuk dipindahkan.

Justin Pihony
sumber
115
Terima kasih atas tanggapannya. Dokumentasi seharusnya lebih baik diucapkan minimize data movementdaripada avoiding data movement.
Praveen Sripati
12
Apakah ada kasus kapan repartitionharus digunakan, bukan coalesce?
Niemand
21
@Niemand Saya pikir dokumentasi saat ini mencakup ini dengan cukup baik: github.com/apache/spark/blob/... Ingatlah bahwa semua yang repartitiondilakukan adalah panggilan coalescedengan shuffleparameter yang disetel ke true. Beri tahu saya jika itu membantu.
Justin Pihony
2
Apakah mungkin untuk mengurangi jumlah file partisi yang ada? Saya tidak punya hdfs, tetapi bermasalah dengan banyak file.
2
partisi ulang akan lebih lambat secara statistik karena tidak tahu bahwa itu menyusut ... walaupun mungkin mereka bisa mengoptimalkannya. Secara internal itu hanya menyebut bersatu dengan shuffle = truebendera
Justin Pihony
171

Jawaban Justin luar biasa dan tanggapan ini masuk lebih dalam.

The repartitionalgoritma tidak shuffle penuh dan menciptakan partisi baru dengan data yang merata. Mari kita membuat DataFrame dengan angka dari 1 hingga 12.

val x = (1 to 12).toList
val numbersDf = x.toDF("number")

numbersDf berisi 4 partisi di mesin saya.

numbersDf.rdd.partitions.size // => 4

Berikut adalah cara membagi data pada partisi:

Partition 00000: 1, 2, 3
Partition 00001: 4, 5, 6
Partition 00002: 7, 8, 9
Partition 00003: 10, 11, 12

Mari kita lakukan shuffle penuh dengan repartitionmetode ini dan dapatkan data ini pada dua node.

val numbersDfR = numbersDf.repartition(2)

Berikut adalah bagaimana numbersDfRdata dipartisi di mesin saya:

Partition A: 1, 3, 4, 6, 7, 9, 10, 12
Partition B: 2, 5, 8, 11

The repartitionMetode membuat partisi baru dan merata mendistribusikan data dalam partisi baru (distribusi data lebih bahkan untuk set data yang lebih besar).

Perbedaan antara coalescedanrepartition

coalescemenggunakan partisi yang ada untuk meminimalkan jumlah data yang dikocok. repartitionmembuat partisi baru dan melakukan pengocokan penuh. coalescemenghasilkan partisi dengan jumlah data yang berbeda (kadang-kadang partisi yang memiliki ukuran yang jauh berbeda) dan repartitionmenghasilkan partisi berukuran kurang lebih sama.

Apakah coalesceatau repartitionlebih cepat?

coalescemungkin berjalan lebih cepat daripada repartition, tetapi partisi berukuran tidak sama umumnya lebih lambat untuk bekerja dengan dari partisi berukuran sama. Anda biasanya perlu mempartisi ulang dataset setelah memfilter kumpulan data besar. Saya menemukan repartitionkeseluruhan lebih cepat karena Spark dibuat untuk bekerja dengan partisi berukuran sama.

NB Saya dengan anehnya mengamati bahwa partisi ulang dapat meningkatkan ukuran data pada disk . Pastikan untuk menjalankan tes saat Anda menggunakan partisi ulang / penggabungan pada kumpulan data besar.

Baca posting blog ini jika Anda ingin lebih detail.

Ketika Anda akan menggunakan gabungan & partisi dalam praktek

Kekuasaan
sumber
8
Jawaban Hebat @Kekuatan, tetapi bukankah data di Partisi A dan B miring? Bagaimana itu didistribusikan secara merata?
anwartheravian
Juga, apa cara terbaik untuk mendapatkan ukuran partisi tanpa mendapatkan kesalahan OOM. Saya menggunakan rdd.glom().map(len).collect()tetapi memberikan banyak kesalahan OOM.
anwartheravian
8
@ anwartheravian - Partisi A dan Partisi B memiliki ukuran yang berbeda karena repartitionalgoritme tidak mendistribusikan data secara merata untuk kumpulan data yang sangat kecil. Saya biasa repartitionmengatur 5 juta catatan menjadi 13 partisi dan setiap file antara 89,3 MB dan 89,6 MB - itu cukup sama!
Powers
1
@Power jawaban ini terlihat lebih baik dengan detail.
Hijau
1
Ini menjelaskan perbedaannya jauh lebih baik. Terima kasih!
Abhi
22

Satu hal tambahan yang perlu diperhatikan di sini adalah, karena prinsip dasar Spark RDD adalah ketetapan. Partisi ulang atau penggabungan akan membuat RDD baru. Basis RDD akan terus memiliki jumlah partisi aslinya. Jika use case menuntut untuk tetap menggunakan RDD dalam cache, maka hal yang sama harus dilakukan untuk RDD yang baru dibuat.

scala> pairMrkt.repartition(10)
res16: org.apache.spark.rdd.RDD[(String, Array[String])] =MapPartitionsRDD[11] at repartition at <console>:26

scala> res16.partitions.length
res17: Int = 10

scala>  pairMrkt.partitions.length
res20: Int = 2
Harikrishnan Ck
sumber
bagus ini penting dan setidaknya untuk scala dev yang berpengalaman ini, tidak jelas - yaitu, baik partisi ulang maupun gabungan upaya untuk memodifikasi data, hanya bagaimana itu didistribusikan di seluruh node
doug
1
@ Harikrishnan jadi jika saya memahami jawaban yang lain dengan benar maka sesuai mereka jika Spark bergabung menggunakan partisi yang ada namun karena RDD tidak dapat diubah, bisakah Anda menggambarkan bagaimana Coalesce memanfaatkan partisi yang ada? Sesuai pemahaman saya, saya pikir Spark menambahkan partisi baru ke partisi yang ada dalam penggabungan.
Explorer
Tetapi jika RDD "lama" tidak digunakan lagi seperti yang diketahui oleh grafik eksekusi, itu akan dihapus dari memori jika tidak bertahan, bukan?
Markus
15

repartition - disarankan untuk menggunakannya sambil meningkatkan jumlah partisi, karena ini melibatkan pengocokan semua data.

coalesce- disarankan untuk menggunakannya sambil mengurangi jumlah partisi. Misalnya jika Anda memiliki 3 partisi dan Anda ingin menguranginya menjadi 2, coalesceakan memindahkan data partisi ke-3 ke partisi 1 dan 2. Partisi 1 dan 2 akan tetap berada di wadah yang sama. Di sisi lain, repartitionakan mengocok data di semua partisi, oleh karena itu penggunaan jaringan antara para pelaksana akan tinggi dan itu akan berdampak pada kinerja.

coalesceberkinerja lebih baik daripada repartitionmengurangi jumlah partisi.

Kamalesan C
sumber
Penjelasan yang Berguna.
Narendra Maru
11

Apa yang mengikuti dari kode dan kode dokumen adalah coalesce(n)sama dengan coalesce(n, shuffle = false)dan repartition(n)sama dengancoalesce(n, shuffle = true)

Dengan demikian, keduanya coalescedan repartitiondapat digunakan untuk menambah jumlah partisi

Dengan shuffle = true, Anda sebenarnya dapat menyatu ke sejumlah besar partisi. Ini berguna jika Anda memiliki sejumlah kecil partisi, katakanlah 100, berpotensi dengan beberapa partisi menjadi besar secara tidak normal.

Catatan penting lain yang perlu ditekankan adalah bahwa jika Anda secara drastis mengurangi jumlah partisi Anda harus mempertimbangkan untuk menggunakan versi shuffledcoalesce (sama seperti repartitiondalam kasus itu). Ini akan memungkinkan perhitungan Anda dilakukan secara paralel pada partisi induk (banyak tugas).

Namun, jika Anda melakukan penggabungan drastis, misalnya ke numPartitions = 1, ini dapat mengakibatkan perhitungan Anda terjadi pada node yang lebih sedikit daripada yang Anda suka (misalnya satu simpul dalam kasus numPartitions = 1). Untuk menghindari ini, Anda bisa lulus shuffle = true. Ini akan menambahkan langkah acak, tetapi berarti partisi hulu saat ini akan dieksekusi secara paralel (per apa pun partisi saat ini).

Lihat juga jawaban terkait di sini

kasur
sumber
10

Semua jawaban menambahkan beberapa pengetahuan hebat ke dalam pertanyaan yang sangat sering ditanyakan ini.

Jadi mengikuti tradisi garis waktu pertanyaan ini, berikut adalah 2 sen saya.

Saya menemukan partisi ulang lebih cepat daripada penggabungan , dalam kasus yang sangat spesifik.

Dalam aplikasi saya ketika jumlah file yang kami perkirakan lebih rendah dari ambang tertentu, partisi ulang bekerja lebih cepat.

Inilah yang saya maksud

if(numFiles > 20)
    df.coalesce(numFiles).write.mode(SaveMode.Overwrite).parquet(dest)
else
    df.repartition(numFiles).write.mode(SaveMode.Overwrite).parquet(dest)

Dalam cuplikan di atas, jika file saya kurang dari 20, penggabungan berlangsung selamanya untuk diselesaikan sementara partisi ulang jauh lebih cepat dan demikian juga kode di atas.

Tentu saja, angka ini (20) akan tergantung pada jumlah pekerja dan jumlah data.

Semoga itu bisa membantu.

Abhishek
sumber
6

Partisi ulang : Kocok data menjadi sejumlah partisi BARU.

Misalnya. Bingkai data awal dipartisi dalam 200 partisi.

df.repartition(500): Data akan diacak dari 200 partisi menjadi 500 partisi baru.

Penggabungan : Kocok data menjadi jumlah partisi yang ada.

df.coalesce(5): Data akan dikocok dari 195 partisi tersisa menjadi 5 partisi yang ada.

Rahul
sumber
3

Saya ingin menambahkan jawaban Justin dan Power bahwa -

repartitionakan mengabaikan partisi yang ada dan membuat yang baru. Jadi Anda bisa menggunakannya untuk memperbaiki kemiringan data. Anda bisa menyebutkan kunci partisi untuk menentukan distribusi. Kemiringan data adalah salah satu masalah terbesar dalam ruang masalah 'data besar'.

coalesceakan bekerja dengan partisi yang ada dan mengacak sebagiannya. Itu tidak dapat memperbaiki kemiringan data sebanyak repartitionhalnya. Karena itu, bahkan jika lebih murah, itu mungkin bukan hal yang Anda butuhkan.

Salim
sumber
3

Untuk semua jawaban hebat yang ingin saya tambahkan itu repartitionadalah salah satu pilihan terbaik untuk memanfaatkan paralelisasi data. Meskipun coalescememberikan opsi yang murah untuk mengurangi partisi dan sangat berguna ketika menulis data ke HDFS atau wastafel lain untuk mengambil keuntungan dari penulisan besar.

Saya menemukan ini berguna ketika menulis data dalam format parket untuk mendapatkan keuntungan penuh.

Ashkrit Sharma
sumber
2

Untuk seseorang yang memiliki masalah menghasilkan file csv tunggal dari PySpark (AWS EMR) sebagai output dan menyimpannya di s3, menggunakan partisi ulang membantu. Alasannya, menyatu tidak bisa melakukan shuffle penuh, tetapi partisi ulang bisa. Pada dasarnya, Anda dapat menambah atau mengurangi jumlah partisi menggunakan partisi ulang, tetapi hanya dapat mengurangi jumlah partisi (tetapi tidak 1) menggunakan gabungan. Ini adalah kode untuk siapa saja yang mencoba menulis csv dari AWS EMR ke s3:

df.repartition(1).write.format('csv')\
.option("path", "s3a://my.bucket.name/location")\
.save(header = 'true')
rampok
sumber
0

Dengan cara sederhana COALESCE: - hanya untuk mengurangi jumlah partisi, Tidak ada pengocokan data, itu hanya mengkompres partisi

REPARTISI: - adalah untuk menambah dan mengurangi no dari partisi, Tetapi terjadi pengocokan

Contoh:-

val rdd = sc.textFile("path",7)
rdd.repartition(10)
rdd.repartition(2)

Keduanya bekerja dengan baik

Tapi kita biasanya pergi untuk dua hal ini ketika kita perlu melihat output dalam satu cluster, kita pergi dengan ini.

Bujuti Niranjan Reddy
sumber
9
Akan ada pergerakan data dengan Coalese juga.
sun_dare
0

Tetapi Anda juga harus memastikan bahwa, data yang akan datang menyatu node harus sangat dikonfigurasi, jika Anda berurusan dengan data besar. Karena semua data akan dimuat ke node tersebut, dapat menyebabkan memori terkecuali. Meskipun reparasi itu mahal, saya lebih suka menggunakannya. Karena itu mengocok dan mendistribusikan data secara merata.

Bijaksana untuk memilih antara penggabungan dan partisi ulang.

Arun Goudar
sumber
0

The repartitionalgoritma melakukan mengocok penuh data dan menciptakan partisi berukuran sama data.coalescemenggabungkan partisi yang ada untuk menghindari pengocokan penuh.

Coalesce bekerja dengan baik untuk mengambil RDD dengan banyak partisi dan menggabungkan partisi pada node pekerja tunggal untuk menghasilkan RDD akhir dengan lebih sedikit partisi.

Repartitionakan merombak data dalam RDD Anda untuk menghasilkan jumlah partisi terakhir yang Anda minta. Pemartisian DataFrames tampak seperti detail implementasi tingkat rendah yang harus dikelola oleh kerangka kerja, tetapi tidak. Saat memfilter DataFrame besar menjadi yang lebih kecil, Anda harus selalu selalu mempartisi data. Anda mungkin akan memfilter DataFrame besar menjadi yang lebih kecil sering, jadi biasakan untuk melakukan partisi ulang.

Baca posting blog ini jika Anda ingin lebih detail.

Sambhav Kumar
sumber