Menjalankan beberapa kueri Streaming Terstruktur pada kluster yang sama

Banyak pelanggan menjalankan beberapa kueri Streaming Terstruktur pada kluster Azure Databricks yang sama. Meskipun pola ini didukung, Databricks merekomendasikan untuk membatasi jumlah kueri per kluster untuk menghindari masalah penskalaan dan penyempitan performa. Pada komputasi tanpa server, Azure Databricks mengelola penskalaan secara otomatis, sehingga pertimbangan ini ditangani untuk Anda. Jika Anda menggunakan komputasi klasik, di mana Anda mengontrol ukuran driver dan eksekutor, halaman ini menjelaskan hambatan kunci yang perlu diingat dan cara untuk mengatasinya.

Note

Databricks merekomendasikan penggunaan alur Lakeflow untuk beban kerja streaming baru, yang mengelola kompleksitas infrastruktur secara otomatis. Lihat Spark Declarative Pipelines.

Kapan menggunakan beberapa kueri pada kluster yang sama

Menjalankan beberapa kueri streaming pada kluster yang sama mengurangi biaya infrastruktur, terutama ketika Anda memiliki banyak aliran kecil yang masing-masing tidak memerlukan komputasi khusus. Tradeoff utama adalah kegagalan bersama: jika kluster gagal, setiap aliran di dalamnya gagal. Untuk pipeline yang sangat kritis, pola kegagalan bersama seperti itu sering kali tidak dapat diterima.

Untuk beban kerja yang mencampur aliran penting dan non-kritis, Databricks merekomendasikan hal berikut:

  • Tetapkan setiap aliran prioritas berdasarkan dampak bisnisnya.
  • Tempatkan aliran misi penting pada kluster khusus, bahkan dengan biaya yang lebih tinggi.
  • Tempatkan stream berprioritas rendah bersama-sama untuk berbagi sumber daya komputasi dan mengurangi biaya.

Penentuan ukuran driver

Driver adalah sumber daya bersama. Beberapa kueri berbagi CPU, memori, penjadwal DAG, penjadwal tugas, dan eksekusi UDF di sisi driver yang sama (misalnya, foreachBatch). Saat menjalankan banyak aliran bersamaan, perhatikan hambatan spesifik ini di luar penyediaan CPU dan memori standar:

  • Overhead pada Auto Loader: Jika stream Anda menggunakan Auto Loader, penemuan file dan pencantuman direktori akan meningkatkan beban pada driver.
  • Batas sumber daya tingkat OS (file terbuka): Menjalankan volume tinggi aliran berbasis file (seperti FileStreamSource atau Auto Loader) secara bersamaan pada satu driver dapat menghabiskan batas deskriptor file terbuka tingkat pengguna, yang dapat menyebabkan kegagalan aliran acak.
  • Backpressure pada bus listener: Banyaknya kueri streaming yang berjalan secara bersamaan dapat menyebabkan backpressure pada bus StreamingQueryListener dari satu sesi Spark. Semua kejadian (termasuk onQueryIdle) dikirim ke bus tunggal ini, dan penumpukan kejadian yang besar dapat menunda secara signifikan penangan onQueryProgress asinkron serta memengaruhi stabilitas kluster.
  • Operasi driver yang mahal: Hindari memanggil collect() atau operasi DataFrame lain yang mahal pada driver kecuali benar-benar diperlukan, untuk menghindari menghasilkan kumpulan hasil yang besar dan menyebabkan kesalahan kehabisan memori (OOM).

Atasi konflik driver

Jika Anda mengalami driver crash karena OOM atau masalah perebutan sumber daya:

  1. Memantau metrik driver di UI Spark. Jika Anda melihat penggunaan CPU, memori, atau disk yang tinggi, sesuaikan ukuran driver dalam pengaturan komputasi kluster.
  2. Jika masalah berlanjut, verifikasi bahwa kode Anda tidak menjalankan operasi intensif memori atau UDF pada driver.
  3. Jika Anda tidak dapat lagi menskalakan driver secara vertikal, Databricks sangat menyarankan agar Anda membagi beban kerja ke beberapa kluster untuk mengatasi kendala penskalaan pada simpul bersama ini.

Ukuran pelaksana

Dengan beberapa kueri yang berjalan pada kluster yang sama, semua kueri berbagi slot tugas pada pelaksana. Tahap-tahap dari satu kueri dapat menempati slot yang tersedia, sehingga menyebabkan penundaan dan terhambatnya eksekusi kueri lain. Spark menggunakan pemetaan 1:1 antara slot tugas dan inti yang tersedia. Pastikan bahwa inti yang cukup tersedia jika kueri perlu dijalankan secara bersamaan.

Secara umum, eksekutor mungkin melakukan lebih banyak operasi intensif memori daripada simpul driver. Sesuaikan JVM eksekutor dan parameter alokasi memori off-heap jika diperlukan untuk menangani beban aplikasi Anda. Pastikan bahwa simpul eksekutor berukuran tepat dalam hal CPU, memori, dan ruang disk dan skala secara vertikal jika diperlukan. Jika penskalakan vertikal tidak dimungkinkan, pertimbangkan untuk menambahkan simpul pekerja tambahan ke kluster.

Note

Beberapa perubahan ini mungkin mengharuskan menghidupkan ulang kluster agar berlaku.

Menggunakan kumpulan penjadwal

Anda dapat mengonfigurasi kumpulan penjadwal untuk menetapkan kapasitas komputasi ke kueri saat menjalankan beberapa kueri streaming dari kode sumber yang sama.

Secara default, semua kueri yang dimulai dalam sebuah notebook dijalankan dalam pool penjadwalan yang adil yang sama. Job Apache Spark yang dipicu dari semua kueri streaming dalam notebook dijalankan satu per satu dalam urutan "yang masuk pertama, keluar pertama" (FIFO). Hal ini dapat menyebabkan penundaan yang tidak perlu pada kueri, karena kueri tersebut tidak menggunakan sumber daya kluster secara bersama dengan efisien.

Kumpulan penjadwal memungkinkan Anda mendeklarasikan kueri Streaming Terstruktur mana yang berbagi sumber daya komputasi.

Contoh berikut menetapkan query1 ke kumpulan khusus, sedangkan query2 dan query3 berbagi kumpulan penjadwal.

:::note Kompatibilitas tanpa server

Databricks menyarankan untuk beralih dari spark.sparkContext karena tidak kompatibel dengan arsitektur komputasi serverless Databricks. Gunakan spark (SparkSession) secara langsung sebagai gantinya. Kumpulan penjadwal adalah konsep komputasi klasik; pada tanpa server, Databricks mengelola penskalaan dan alokasi sumber daya secara otomatis.

:::

# Run streaming query1 in scheduler pool1
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
df.writeStream.queryName("query1").toTable("table1")

# Run streaming query2 in scheduler pool2
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
df.writeStream.queryName("query2").toTable("table2")

# Run streaming query3 in scheduler pool2
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
df.writeStream.queryName("query3").toTable("table3")

Note

Konfigurasi properti lokal harus berada di sel buku catatan yang sama tempat Anda memulai kueri streaming.

Untuk informasi selengkapnya tentang kumpulan penjadwal yang adil, lihat dokumentasi penjadwal adil Apache Spark.

Pertimbangan untuk kueri stateful

Untuk kueri stateful yang berjalan pada kluster yang sama, ingatlah hal berikut:

  • Gunakan RocksDB sebagai penyedia penyimpanan status untuk menghindari masalah OOM dan jeda GC. RocksDB adalah penyedia penyimpanan status default di Databricks Runtime 17.3 ke atas. Lihat Mengonfigurasi penyimpanan status RocksDB di Azure Databricks.
  • Sesuaikan partisi shuffle sesuai kebutuhan aplikasi Anda. Untuk tahapan stateful, Spark menjadwalkan tugas yang sebanding dengan jumlah partisi shuffle.
  • Batasi penggunaan memori RocksDB per simpul untuk menghindari error OOM akibat penggunaan memori off-heap. Ini ditangani secara otomatis di Databricks Runtime 17.3 ke atas, tetapi memerlukan konfigurasi manual pada rilis sebelumnya. Lihat Penggunaan memori Cap RocksDB.
  • Hindari mengemas terlalu banyak partisi pada node eksekutor yang sama. Operasi pemeliharaan pada penyimpanan status, termasuk pengunggahan snapshot dan pembersihannya, dijalankan per simpul. Menetapkan terlalu banyak partisi ke satu simpul eksekutor dapat menyebabkan kelaparan pemeliharaan dan waktu pemulihan yang lebih lama karena lebih sedikit rekam jepret penuh yang tersedia.
  • Last updated on