Tips penyesuaian tugas Spark

Bagian berikut memberikan tips untuk membantu Anda menyesuaikan aplikasi Spark Managed Service untuk Apache Spark.

Menggunakan cluster sementara

Saat menggunakan model cluster "efemeral" Managed Service untuk Apache Spark, Anda membuat cluster khusus untuk setiap tugas, dan saat tugas selesai, Anda menghapus cluster tersebut. Dengan model sementara, Anda dapat memperlakukan penyimpanan dan komputasi secara terpisah, menyimpan data input dan output tugas di Cloud Storage atau BigQuery, menggunakan cluster hanya untuk komputasi dan penyimpanan data sementara.

Kesalahan umum cluster persisten

Penggunaan cluster satu tugas efemeral menghindari perangkap dan potensi masalah berikut yang terkait dengan penggunaan cluster "persisten" bersama dan yang berjalan lama:

  • Titik tunggal kegagalan: status error cluster bersama dapat menyebabkan semua tugas gagal, sehingga memblokir seluruh pipeline data. Menyelidiki dan memulihkan dari error bisa memakan waktu berjam-jam. Karena cluster sementara hanya menyimpan status dalam cluster sementara, saat terjadi error, cluster tersebut dapat dihapus dan dibuat ulang dengan cepat.
  • Kesulitan dalam memelihara dan memigrasikan status cluster di HDFS, MySQL, atau sistem file lokal
  • Pertentangan resource di antara tugas yang berdampak negatif pada SLO
  • Daemon layanan tidak responsif yang disebabkan oleh tekanan memori
  • Penumpukan log dan file sementara yang dapat melebihi kapasitas disk
  • Kegagalan peningkatan skala karena kehabisan stok zona cluster
  • Kurangnya dukungan untuk versi image cluster yang sudah tidak berlaku.

Manfaat cluster sementara

Di sisi positif, cluster efemeral memungkinkan Anda melakukan hal berikut:

  • Konfigurasikan izin IAM yang berbeda untuk tugas yang berbeda dengan akun layanan VM Managed Service untuk Apache Spark yang berbeda.
  • Mengoptimalkan konfigurasi hardware dan software cluster untuk setiap tugas, mengubah konfigurasi cluster sesuai kebutuhan.
  • Upgrade versi image di cluster baru untuk mendapatkan patch keamanan, perbaikan bug, dan pengoptimalan terbaru.
  • Memecahkan masalah dengan lebih cepat di cluster tugas tunggal yang terisolasi.
  • Hemat biaya dengan hanya membayar waktu berjalan cluster sementara, bukan untuk waktu tidak ada aktivitas di antara tugas pada cluster bersama.

Menggunakan Spark SQL

DataFrame API Spark SQL merupakan pengoptimalan signifikan dari RDD API. Jika Anda berinteraksi dengan kode yang menggunakan RDD, pertimbangkan untuk membaca data sebagai DataFrame sebelum meneruskan RDD dalam kode. Dalam kode Java atau Scala, pertimbangkan untuk menggunakan Dataset API Spark SQL sebagai superset RDD dan DataFrame.

Menggunakan Apache Spark 3

Managed Service untuk Apache Spark 2.0 menginstal Spark 3, yang mencakup fitur dan peningkatan performa berikut:

  • Dukungan GPU
  • Kemampuan untuk membaca file biner
  • Peningkatan performa
  • Pemangkasan Partisi Dinamis
  • Eksekusi kueri adaptif, yang mengoptimalkan tugas Spark secara real time

Menggunakan Alokasi Dinamis

Apache Spark menyertakan fitur Alokasi Dinamis yang menskalakan jumlah eksekutor Spark pada worker dalam cluster. Fitur ini memungkinkan tugas menggunakan cluster Managed Service untuk Apache Spark secara penuh, bahkan saat cluster melakukan penskalaan. Fitur ini diaktifkan secara default di Managed Service untuk Apache Spark (spark.dynamicAllocation.enabled ditetapkan ke true). Lihat Alokasi Dinamis Spark untuk mengetahui informasi selengkapnya.

Menggunakan penskalaan otomatis Managed Service untuk Apache Spark

Autoscaling Managed Service untuk Apache Spark menambahkan dan menghapus worker Managed Service untuk Apache Spark secara dinamis dari cluster untuk membantu memastikan bahwa tugas Spark memiliki resource yang diperlukan untuk diselesaikan dengan cepat.

Sebaiknya praktik terbaik mengonfigurasi kebijakan penskalaan otomatis agar hanya menskalakan worker sekunder.

Menggunakan Mode Fleksibilitas yang Ditingkatkan di Managed Service untuk Apache Spark

Cluster dengan preemptible VM atau kebijakan penskalaan otomatis dapat menerima pengecualian FetchFailed saat worker di-preempt atau dihapus sebelum selesai menyajikan data acak ke reducer. Pengecualian ini dapat menyebabkan percobaan ulang tugas dan waktu penyelesaian tugas yang lebih lama.

Rekomendasi: Gunakan Mode Fleksibilitas yang Ditingkatkan Managed Service untuk Apache Spark, yang tidak menyimpan data shuffle perantara di worker sekunder, sehingga worker sekunder dapat di-preempt atau diperkecil skalanya dengan aman.

Mengonfigurasi partisi dan pengacakan

Spark menyimpan data dalam partisi sementara di cluster. Jika aplikasi Anda mengelompokkan atau menggabungkan DataFrame, aplikasi akan mengacak data ke dalam partisi baru sesuai dengan pengelompokan dan konfigurasi tingkat rendah.

Partisi data berdampak signifikan pada performa aplikasi: partisi yang terlalu sedikit akan membatasi paralelisme tugas dan pemanfaatan resource cluster; partisi yang terlalu banyak akan memperlambat tugas karena pemrosesan dan pengacakan partisi tambahan.

Mengonfigurasi partisi

Properti berikut mengatur jumlah dan ukuran partisi Anda:

  • spark.sql.files.maxPartitionBytes: ukuran maksimum partisi saat Anda membaca data dari Cloud Storage. Defaultnya adalah 128 MB, yang cukup besar untuk sebagian besar aplikasi yang memproses kurang dari 100 TB.

  • spark.sql.shuffle.partitions: jumlah partisi setelah melakukan shuffle. Defaultnya adalah 1000 untuk cluster versi image 2.2 dan yang lebih baru. Rekomendasi: Tetapkan ini ke 3x jumlah vCPU di cluster Anda.

  • spark.default.parallelism: jumlah partisi yang ditampilkan setelah melakukan transformasi RDD yang memerlukan shuffle, seperti join, reduceByKey, dan parallelize. Defaultnya adalah jumlah total vCPU di cluster Anda. Saat menggunakan RDD dalam tugas Spark, Anda dapat menetapkan angka ini menjadi 3x vCPU Anda

Membatasi jumlah file

Ada penurunan performa saat Spark membaca file kecil dalam jumlah besar. Simpan data dalam ukuran file yang lebih besar, misalnya, ukuran file dalam rentang 256 MB–512 MB. Demikian pula, batasi jumlah file output (untuk memaksa pengacakan, lihat Menghindari pengacakan yang tidak perlu).

Mengonfigurasi adaptive query execution (Spark 3)

Eksekusi kueri adaptif (diaktifkan secara default di Managed Service untuk Apache Spark versi image 2.0) memberikan peningkatan performa tugas Spark, termasuk:

Meskipun setelan konfigurasi default sudah sesuai untuk sebagian besar kasus penggunaan, menyetel spark.sql.adaptive.advisoryPartitionSizeInBytes ke spark.sql.files.maxPartitionBytes (default 128 MB) dapat bermanfaat.

Menghindari pengacakan yang tidak perlu

Spark memungkinkan pengguna memicu shuffle secara manual untuk menyeimbangkan kembali data mereka dengan fungsi repartition. Pengacakan memerlukan banyak biaya, jadi pengacakan ulang data harus digunakan dengan hati-hati. Menetapkan konfigurasi partisi dengan tepat seharusnya sudah cukup untuk memungkinkan Spark mempartisi data Anda secara otomatis.

Pengecualian: Saat menulis data yang dipartisi kolom ke Cloud Storage, pemartisian ulang pada kolom tertentu akan menghindari penulisan banyak file kecil untuk mencapai waktu penulisan yang lebih cepat. 

df.repartition("col_name").write().partitionBy("col_name").save("gs://...")

Menyimpan data di Parquet atau Avro

Spark SQL secara default membaca dan menulis data dalam file Parquet yang dikompresi Snappy. Parquet adalah format file berbasis kolom yang efisien, yang memungkinkan Spark hanya membaca data yang diperlukan untuk menjalankan aplikasi. Hal ini merupakan keuntungan penting saat bekerja dengan set data besar. Format berbasis kolom lainnya, seperti Apache ORC, juga berperforma baik.

Untuk data non-kolom, Apache Avro menyediakan format file baris biner yang efisien. Meskipun biasanya lebih lambat daripada Parquet, performa Avro lebih baik daripada format berbasis teks, seperti CSV atau JSON.

Mengoptimalkan ukuran disk

Throughput persistent disk diskalakan sesuai ukuran disk, yang dapat memengaruhi performa tugas Spark karena tugas menulis metadata dan mengacak data ke disk. Saat menggunakan persistent disk standar, ukuran disk harus minimal 1 terabyte per worker (lihat Performa berdasarkan ukuran persistent disk).

Untuk memantau throughput disk worker di konsolGoogle Cloud :

  1. Klik nama cluster di halaman Clusters.
  2. Klik tab VM INSTANCES.
  3. Klik nama pekerja mana pun.
  4. Klik tab MONITORING, lalu scroll ke Disk Throughput untuk melihat throughput worker.

Pertimbangan disk

Cluster Managed Service untuk Apache Spark sementara, yang tidak memanfaatkan penyimpanan persisten, dapat menggunakan SSD lokal. SSD lokal secara fisik terpasang ke cluster dan memberikan throughput yang lebih tinggi daripada persistent disk (lihat Tabel performa). SSD lokal tersedia dalam ukuran tetap sebesar 375 gigabyte, tetapi Anda dapat menambahkan beberapa SSD untuk meningkatkan performa.

SSD lokal tidak mempertahankan data setelah cluster dinonaktifkan. Jika memerlukan penyimpanan persisten, Anda dapat menggunakan persistent disk SSD, yang memberikan throughput lebih tinggi untuk ukurannya dibandingkan persistent disk standar. Persistent disk SSD juga merupakan pilihan yang baik jika ukuran partisi akan lebih kecil dari 8 KB (namun, hindari partisi kecil).

Memasang GPU ke cluster Anda

Spark 3 mendukung GPU. Gunakan GPU dengan tindakan inisialisasi CUDA-X untuk mempercepat tugas Spark menggunakan library NVIDIA cuDF untuk Apache Spark. Tindakan inisialisasi driver GPU untuk mengonfigurasi cluster dengan GPU.

Kegagalan dan perbaikan tugas umum

Bagian berikut membahas penyebab umum dan solusi untuk kegagalan tugas.

Kehabisan Memori

Contoh:

  • "Lost executor"
  • "java.lang.OutOfMemoryError: GC overhead limit exceeded"
  • "Container dihentikan oleh YARN karena melebihi batas memori"

Kemungkinan perbaikan:

Kegagalan Pengambilan Shuffle

Contoh:

  • "FetchFailedException" (error Spark)
  • "Gagal terhubung ke..." (Error Spark)
  • "Failed to fetch" (error MapReduce)

Biasanya disebabkan oleh penghapusan worker secara prematur yang masih memiliki data shuffle untuk disajikan.

Kemungkinan penyebab dan perbaikan:

  • Worker VM preemptible diklaim kembali atau worker VM non-preemptible dihapus oleh autoscaler. Solusi: Gunakan Mode Fleksibilitas yang Ditingkatkan untuk membuat worker sekunder dapat di-preempt atau diskalakan dengan aman.
  • Executor atau mapper error karena error OutOfMemory. Solusi: tingkatkan memori eksekutor atau mapper.
  • Layanan shuffle Spark mungkin kelebihan beban. Solusi: kurangi jumlah partisi tugas.

Node YARN TIDAK SEHAT

Contoh (dari log YARN): 

...reported UNHEALTHY with details: 1/1 local-dirs usable space is below
configured utilization percentage/no more usable space
[ /hadoop/yarn/nm-local-dir : used space above threshold of 90.0% ]

Sering kali terkait dengan kapasitas disk yang tidak memadai untuk data shuffle. Diagnosis dengan melihat file log:

  • Buka halaman Clusters project Anda di konsol Google Cloud , lalu klik nama cluster.
  • Klik VIEW LOGS.
  • Filter log menurut hadoop-yarn-nodemanager.
  • Telusuri "UNHEALTHY".

Kemungkinan Perbaikan:

  • Cache pengguna disimpan di direktori yang ditentukan oleh properti yarn.nodemanager.local-dirs di yarn-site.xml file. File ini terletak di /etc/hadoop/conf/yarn-site.xml. Anda dapat memeriksa ruang kosong di jalur /hadoop/yarn/nm-local-dir, dan mengosongkan ruang dengan menghapus folder cache pengguna /hadoop/yarn/nm-local-dir/usercache.
  • Jika log melaporkan status "UNHEALTHY", buat ulang cluster Anda dengan kapasitas disk yang lebih besar, yang akan meningkatkan batas throughput.

Cache pengguna menghabiskan kapasitas disk

Di Managed Service untuk Apache Spark, cache pengguna disimpan di direktori yang ditentukan oleh properti yarn.nodemanager.local-dirs di yarn-site.xml (/etc/hadoop/conf/yarn-site.xml). Biasanya, lokasinya adalah /hadoop/yarn/nm-local-dir.

Kemungkinan Perbaikan:

  • Pembersihan manual: Periksa ruang kosong di jalur /hadoop/yarn/nm-local-dir, lalu kosongkan ruang dengan menghapus folder cache pengguna /hadoop/yarn/nm-local-dir/usercache.

  • Penyelesaian jangka panjang: Kelola proses pembersihan dengan menetapkan properti cluster berikut:

    • yarn.nodemanager.localizer.cache.cleanup.interval-ms
    • yarn.nodemanager.localizer.cache.target-size-mb

    Jika tidak ingin membuat ulang cluster, Anda dapat memperbarui properti ini di yarn-site.xml pada semua node, lalu memulai ulang NodeManager dengan menjalankan perintah berikut pada worker node:

    sudo systemctl restart hadoop-yarn-nodemanager.service

  • Ubah ukuran cluster: Jika log melaporkan status "UNHEALTHY", buat ulang cluster Anda dengan kapasitas disk yang lebih besar, yang akan meningkatkan batas throughput.

Tugas gagal karena memori driver tidak mencukupi

Saat menjalankan tugas dalam mode cluster, tugas akan gagal jika ukuran memori worker node melebihi ukuran memori worker node.

Contoh dari log driver: 

'Exception in thread "main" java.lang.IllegalArgumentException:
Required AM memory (32768+3276 MB) is above the max threshold (12288 MB) of this cluster!
Please check the values of 'yarn.scheduler.maximum -allocation-mb' and/or 'yarn.nodemanager.resource.memory-mb'.'

Kemungkinan Perbaikan:

  • Tetapkan spark:spark.driver.memory kurang dari yarn:yarn.scheduler.maximum-allocation-mb.
  • Gunakan jenis mesin yang sama untuk node master dan worker.

Langkah berikutnya