Mari kita asumsikan sebagai berikut bahwa hanya satu tugas Spark yang berjalan di setiap titik waktu.

Apa yang saya dapatkan sejauh ini

Inilah yang saya pahami tentang apa yang terjadi di Spark:

1. Saat a SparkContextdibuat, setiap node pekerja memulai eksekutor. Pelaksana adalah proses terpisah (JVM), yang menghubungkan kembali ke program driver. Setiap pelaksana memiliki toples program driver. Keluar dari pengemudi, matikan pelaksana. Setiap pelaksana dapat menampung beberapa partisi.
2. Ketika sebuah pekerjaan dijalankan, rencana eksekusi dibuat sesuai dengan grafik garis keturunan.
3. Pekerjaan eksekusi dibagi menjadi beberapa tahap, di mana tahapan yang berisi transformasi dan tindakan tetangga (dalam grafik garis keturunan) sebanyak mungkin, tetapi tidak ada pengacakan. Jadi tahapan dipisahkan oleh pengocokan.

aku mengerti itu

• Tugas adalah perintah yang dikirim dari driver ke eksekutor dengan membuat serial objek Fungsi.
• Eksekutor deserializes (dengan driver jar) perintah (tugas) dan mengeksekusinya di partisi.

tapi

Pertanyaan

Bagaimana cara membagi tahapan menjadi tugas-tugas itu?

Secara khusus:

1. Apakah tugas ditentukan oleh transformasi dan tindakan atau dapatkah beberapa transformasi / tindakan menjadi tugas?
2. Apakah tugas ditentukan oleh partisi (misalnya satu tugas per tahap per partisi).
3. Apakah tugas ditentukan oleh node (misalnya satu tugas per tahap per node)?

Apa yang saya pikirkan (hanya jawaban parsial, meskipun benar)

Di https://0x0fff.com/spark-architecture-shuffle , pengocokan dijelaskan dengan gambar

dan saya mendapat kesan bahwa aturannya adalah

setiap tahap dibagi menjadi # tugas jumlah partisi, tanpa memperhatikan jumlah node

Untuk gambar pertama saya, saya akan mengatakan bahwa saya memiliki 3 tugas peta dan 3 tugas pengurangan.

Untuk gambar dari 0x0fff, saya akan mengatakan ada 8 tugas peta dan 3 tugas pengurangan (dengan asumsi bahwa hanya ada tiga file oranye dan tiga file hijau tua).

Pertanyaan terbuka dalam hal apa pun

Apakah itu benar? Tetapi meskipun itu benar, pertanyaan saya di atas tidak semuanya terjawab, karena masih terbuka, apakah beberapa operasi (mis. Beberapa peta) berada dalam satu tugas atau dipisahkan menjadi satu tugas per operasi.

Apa yang dikatakan orang lain

Apa tugas di Spark? Bagaimana cara pekerja Spark mengeksekusi file jar? dan Bagaimana penjadwal Apache Spark membagi file menjadi tugas? mirip, tetapi saya tidak merasa pertanyaan saya dijawab dengan jelas di sana.

Anda memiliki garis besar yang cukup bagus di sini. Untuk menjawab pertanyaan Anda

• Sebuah terpisah task tidak perlu diluncurkan untuk setiap partisi data untuk masing-masing stage. Pertimbangkan bahwa setiap partisi kemungkinan besar akan berada di lokasi fisik yang berbeda - misalnya blok di HDFS atau direktori / volume untuk sistem file lokal.

Perhatikan bahwa pengiriman Stages didorong oleh DAG Scheduler. Ini berarti bahwa tahapan yang tidak saling bergantung dapat dikirimkan ke cluster untuk dieksekusi secara paralel: ini memaksimalkan kemampuan paralelisasi pada cluster. Jadi, jika operasi dalam aliran data kami dapat terjadi secara bersamaan, kami berharap akan melihat beberapa tahapan diluncurkan.

Kita dapat melihat itu beraksi dalam contoh mainan berikut di mana kita melakukan jenis operasi berikut:

• memuat dua sumber data
• melakukan beberapa operasi peta pada kedua sumber data secara terpisah
• bergabung dengan mereka
• melakukan beberapa operasi peta dan filter pada hasilnya
• simpan hasilnya

Jadi, berapa banyak tahapan yang akan kita capai?

• Masing-masing 1 tahap untuk memuat dua sumber data secara paralel = 2 tahap
• Tahap ketiga yang mewakili joinyang bergantung pada dua tahap lainnya
• Catatan: semua operasi lanjutan yang bekerja pada data yang digabungkan dapat dilakukan dalam tahap yang sama karena harus terjadi secara berurutan. Tidak ada manfaat untuk meluncurkan tahapan tambahan karena mereka tidak dapat mulai bekerja sampai operasi sebelumnya diselesaikan.

Ini program mainan itu

val sfi  = sc.textFile("/data/blah/input").map{ x => val xi = x.toInt; (xi,xi*xi) }
val sp = sc.parallelize{ (0 until 1000).map{ x => (x,x * x+1) }}
val spj = sfi.join(sp)
val sm = spj.mapPartitions{ iter => iter.map{ case (k,(v1,v2)) => (k, v1+v2) }}
val sf = sm.filter{ case (k,v) => v % 10 == 0 }
sf.saveAsTextFile("/data/blah/out")

Dan inilah DAG dari hasilnya

Sekarang: berapa banyak tugas ? Jumlah tugas harus sama dengan

Jumlah ( Stage* #Partitions in the stage)

Ini mungkin membantu Anda lebih memahami bagian yang berbeda:

• Tahap: adalah kumpulan tugas. Proses yang sama berjalan pada subset data (partisi) yang berbeda.
• Tugas: merepresentasikan unit kerja pada partisi dari kumpulan data terdistribusi. Jadi di setiap tahap, jumlah-tugas = jumlah-partisi, atau seperti yang Anda katakan "satu tugas per tahap per partisi".
• Setiap pelaksana berjalan pada satu wadah benang, dan setiap wadah berada pada satu simpul.
• Setiap tahap menggunakan beberapa eksekutor, setiap eksekusi dialokasikan pada beberapa vcores.
• Setiap vcore dapat menjalankan tepat satu tugas dalam satu waktu
• Jadi pada tahap mana pun, banyak tugas dapat dijalankan secara paralel. number-of-task running = jumlah-of-vcores yang digunakan.

Jika saya mengerti dengan benar ada 2 hal (terkait) yang membingungkan Anda:

1) Apa yang menentukan konten tugas?

2) Apa yang menentukan jumlah tugas yang akan dijalankan?

Mesin Spark "merekatkan" operasi sederhana pada rdds yang berurutan, misalnya:

rdd1 = sc.textFile( ... )
rdd2 = rdd1.filter( ... )
rdd3 = rdd2.map( ... )
rdd3RowCount = rdd3.count

jadi ketika rdd3 (malas) dihitung, spark akan menghasilkan tugas per partisi rdd1 dan setiap tugas akan mengeksekusi filter dan peta per baris untuk menghasilkan rdd3.

Jumlah tugas ditentukan oleh jumlah partisi. Setiap RDD memiliki jumlah partisi yang ditentukan. Untuk RDD sumber yang dibaca dari HDFS (misalnya menggunakan sc.textFile (...)) jumlah partisi adalah jumlah pemisahan yang dihasilkan oleh format input. Beberapa operasi pada RDD dapat menghasilkan RDD dengan jumlah partisi yang berbeda:

rdd2 = rdd1.repartition( 1000 ) will result in rdd2 having 1000 partitions ( regardless of how many partitions rdd1 had ).

Contoh lainnya adalah join:

rdd3 = rdd1.join( rdd2  , numPartitions = 1000 ) will result in rdd3 having 1000 partitions ( regardless of partitions number of rdd1 and rdd2 ).

(Sebagian besar) operasi yang mengubah jumlah partisi melibatkan pengacakan, Ketika kami melakukannya misalnya:

rdd2 = rdd1.repartition( 1000 ) 

yang sebenarnya terjadi adalah tugas pada setiap partisi rdd1 perlu menghasilkan keluaran akhir yang dapat dibaca oleh tahap berikut sehingga untuk membuat rdd2 memiliki tepat 1000 partisi (Bagaimana mereka melakukannya? Hash atau Urutkan ). Tugas di sisi ini kadang-kadang disebut sebagai "Tugas peta (samping)". Sebuah tugas yang nantinya akan berjalan di rdd2 akan bekerja pada satu partisi (dari rdd2!) Dan harus mencari cara untuk membaca / menggabungkan keluaran sisi peta yang relevan dengan partisi tersebut. Tugas di sisi ini terkadang disebut sebagai "Kurangi (tugas sampingan)".

2 pertanyaan terkait: jumlah tugas dalam sebuah tahapan adalah jumlah partisi (umum untuk rdds berurutan yang "direkatkan" bersama) dan jumlah partisi rdd dapat berubah antar tahapan (dengan menentukan jumlah partisi ke beberapa shuffle menyebabkan operasi misalnya).

Setelah eksekusi tahapan dimulai, tugasnya dapat menempati slot tugas. Jumlah slot tugas serentak adalah numExecutors * ExecutorCores. Secara umum, ini dapat ditempati oleh tugas-tugas dari tahapan yang berbeda dan tidak bergantung.