KAL
Menu

Job, stage, task

DAG pekerjaan dan dependensi wide versus narrow.

Daftar isi

Satu action pada Spark memicu satu job, dan job terpecah menjadi stage yang masing-masing berisi sekumpulan task paralel. Topik ini membahas cara job disusun dari graf transformasi, aturan pemecahan stage berdasarkan jenis dependensi antar simpulnya, dan peran task sebagai satuan kerja terkecil yang dijadwalkan pada satu partisi. Setelah membacanya, isi Spark UI dapat dibaca dengan benar, dan jumlah stage suatu pekerjaan dapat diprediksi dari bentuk kodenya.

Dari action menjadi job dan DAG#

Ketika action dipanggil, Spark menyusun DAG, yaitu graf asiklik terarah dari seluruh transformasi yang menyusun hasilnya. Graf itu adalah representasi lengkap pekerjaan, mulai dari simpul sumber data sampai simpul hasil, lengkap dengan operator yang diperlukan di setiap sambungan. Direncanakan pada level SQL, graf ini lalu dieksekusi sesuai urutan dependensinya, bukan sesuai urutan penulisan kode, sehingga dua cabang yang tidak saling bergantung dapat dieksekusi paralel.

Satu action menghasilkan satu job, dan banyak program nyata berisi banyak job karena banyak action. Pada Spark UI, setiap job tercatat dengan durasinya dan stage-stagenya, sehingga program yang terasa lambat dapat dilokalkan ke job tertentu terlebih dahulu sebelum penyebabnya dicari. Kebiasaan memberi nama aplikasi yang bermakna pada appName membantu saat daftar job ini sudah panjang.

Wide versus narrow: batas pemecah stage#

Aturan pemecahan stage terletak pada jenis dependensi antar simpul DAG. Dependensi narrow menyatakan setiap partisi keluaran cukup dihitung dari satu partisi masukan, seperti filter, select, dan withColumn, sehingga task-nya dapat berjalan tanpa menunggu partisi lain. Dependensi wide menyatakan satu partisi keluaran memerlukan gabungan data dari banyak partisi masukan, seperti groupBy, orderBy, dan join, karena kunci kelompok yang sama tersebar di berbagai partisi.

Batas antara stage adalah tepat di dependensi wide, dan di batas itulah data ditukar antar pekerja melalui proses shuffle. Keluaran explain() pada topik ini memperlihatkan struktur tersebut secara langsung, yaitu rantai operator yang terkumpul dalam satu stage lalu terpotong oleh simpul Exchange.

python peraga_rencana.py 
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("peraga-rencana")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

df = spark.createDataFrame(
    [(1, "Ali", "10A", 85), (2, "Budi", "10A", 90), (3, "Citra", "10B", 78), (4, "Dewi", "10B", 92)],
    "id long, nama string, kelas string, nilai long",
)

rencana = (
    df.filter(F.col("nilai") >= 80)
    .groupBy("kelas")
    .agg(F.avg("nilai").alias("rata_rata"))
)
rencana.explain()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[kelas#2], functions=[avg(nilai#3L)])
   +- Exchange hashpartitioning(kelas#2, 2), ENSURE_REQUIREMENTS, [plan_id=19]
      +- HashAggregate(keys=[kelas#2], functions=[partial_avg(nilai#3L)])
         +- Project [kelas#2, nilai#3L]
            +- Filter (isnotnull(nilai#3L) AND (nilai#3L >= 80))
               +- Scan ExistingRDD[id#0L,nama#1,kelas#2,nilai#3L]

Pembacaan rencana dilakukan dari bawah ke atas. Operator Scan dan Filter serta Project berada pada stage pertama yang dependensinya narrow, kemudian Exchange hashpartitioning menandai batas shuffle ke stage kedua. Kemunculan dua lapis HashAggregate adalah hasil pembagian agregasi menjadi tahap partial pada tiap partisi dan tahap akhir lintas partisi, sehingga sebagian pengurangan data sudah terjadi sebelum pertukaran. Tanda AdaptiveSparkPlan menunjukkan AQE (Adaptive Query Execution) aktif dan rencana final belum dikunci, sesuai bahasan topik AQE pada modul produksi.

Task: satuan kerja pada satu partisi#

Stage dieksekusi sebagai kumpulan task, dan setiap task mengerjakan satu partisi data dengan operator yang sama. Jumlah task sebuah stage mengikuti jumlah partisi pada input stage itu, sehingga jumlah partisi menjadi penentu langsung paralelisme. Task dijadwalkan ke slot eksekusi yang tersedia, dan pada mode lokal local[2] jumlah slotnya dua thread, sehingga task yang lebih banyak dari itu mengantre di dalam proses yang sama.

Kaitan antara jumlah task dan partisi inilah yang menjelaskan banyak perilaku yang sebelumnya tampak ajaib. Data empat baris yang terbagi dua partisi menghasilkan dua task pada stage baca, sementara agregasi dengan spark.sql.shuffle.partitions=2 menghasilkan dua task pada stage berikutnya. Pada data besar, jumlah task per stage bisa ribuan, dan ukuran task yang terlalu kecil justru memakan biaya penjadwalan, tema yang diulang pada topik partisi.

Membaca job, stage, dan task juga melatih kebiasaan menghubungkan angka dengan transformasi. Satu pekerjaan sederhana bisa memuat beberapa stage karena tiap shuffle memisahkan stage, dan jumlah task per stage mengikuti jumlah partisi sisi masing-masing. Ketika sebuah pekerjaan tampak membutuhkan waktu lama, pertanyaan pertama yang berguna adalah stage mana yang paling lama dan berapa task yang menyusunnya, karena jawabannya langsung mengarah ke transformasi yang menimbulkan shuffle. Pola membaca inilah yang dipakai pada modul produksi untuk mendiagnosis pekerjaan nyata melalui Spark UI.

Kesalahan umum#

Kesalahan pertama adalah mengira jumlah stage sama dengan jumlah transformasi. Stage dibentuk oleh batas dependensi wide, dan belasan transformasi narrow tetap satu stage penuh. Prediksi yang benar adalah menghitung Exchange pada rencana, bukan menghitung baris kode.

Kesalahan kedua adalah membaca rencana explain() dari atas ke bawah. Struktur rencana dicetak dengan induk di atas dan sumber data di bawah, sehingga alur data mengikuti arah bawah-ke-atas. Salah arah pembacaan membuat interpretasi filter dan agregasi menjadi terbalik.

Kesalahan ketiga adalah menganggap semua task sama besar. Partisi yang isinya timpang menghasilkan task yang timpang, dan stage hanya selesai ketika task terlamannya selesai, fenomena skew yang dibahas lengkap pada topik partisi. Gejalanya pada Spark UI adalah satu task bertahan jauh lebih lama daripada yang lain pada stage yang sama.

Ringkasan#

  • Satu action menghasilkan satu job yang direncanakan sebagai DAG transformasi, dan DAG itu dieksekusi mengikuti dependensi, bukan urutan penulisan kode.
  • Dependensi narrow mempertahankan partisi dan tidak memecah stage, sedangkan dependensi wide memicu shuffle dan menjadi batas stage.
  • explain() mencetak rencana fisik dari bawah ke atas, dengan Exchange sebagai penanda batas stage dan HashAggregate ganda sebagai tanda agregasi bertahap.
  • Task adalah satuan kerja satu partisi, sehingga jumlah dan ukuran partisi menentukan jumlah dan keseimbangan task.

Referensi#