Streaming Terstruktur pertama kali diperkenalkan di Apache Spark 2.0. Platform ini telah memantapkan dirinya sebagai pilihan terbaik untuk membangun aplikasi streaming terdistribusi. Penyatuan SQL / Dataset / DataFrame API dan fungsi Spark bawaan membuatnya lebih mudah bagi developer untuk mengimplementasikan hal-hal penting yang kompleks seperti streaming aggregation, stream-stream join, dan dukungan windowing. Sejak rilis Streaming Terstruktur, telah menjadi permintaan populer dari pengembang untuk meningkatkan kontrol streaming, seperti yang kami lakukan di Spark Streaming (seperti DStream). Di Apache Spark 3.0, kami merilis UI baru untuk Streaming Terstruktur.
Streaming Terstruktur UI baru menyediakan cara mudah untuk memantau semua tugas streaming dengan wawasan dan statistik yang dapat ditindaklanjuti, membuatnya lebih mudah untuk memecahkan masalah saat melakukan debug, dan meningkatkan visibilitas produksi dengan metrik waktu nyata. UI menyajikan dua set statistik: 1) informasi agregat tentang pekerjaan kueri streaming dan 2) informasi statistik terperinci tentang permintaan streaming, termasuk Tingkat Input, Kecepatan Proses, Baris Input, Durasi Batch, Durasi Operasi, dll.
Informasi yang dikumpulkan tentang pekerjaan kueri streaming
Saat pengembang mengirimkan kueri SQL streaming, kueri tersebut akan muncul di tab Streaming Terstruktur, yang menyertakan kueri streaming aktif dan yang sudah selesai. Tabel hasil akan memberikan beberapa informasi dasar mengenai permintaan streaming, termasuk nama permintaan, status, ID, runID, waktu pengiriman, durasi permintaan, ID paket terakhir, serta informasi agregat seperti rata-rata tingkat penerimaan dan kecepatan pemrosesan rata-rata. Ada tiga jenis status permintaan streaming: MENJALANKAN, SELESAI, dan GAGAL. Semua permintaan SELESAI dan GAGAL dicantumkan di tabel permintaan streaming lengkap. Kolom Error menampilkan detail dari pengecualian permintaan yang gagal.
Kami dapat melihat statistik rinci dari permintaan streaming dengan mengklik link ID Jalankan.
Informasi statistik terperinci
Halaman Statistik menampilkan metrik termasuk laju penyerapan / pemrosesan, latensi, dan durasi operasi mendetail, yang berguna untuk memahami status permintaan streaming Anda, sehingga memudahkan untuk men-debug anomali dalam pemrosesan permintaan.
Ini berisi metrik berikut:
- Tingkat Input : tingkat kedatangan data gabungan (di semua sumber).
- Kecepatan Proses : Tingkat gabungan (di semua sumber) di mana Spark memproses data.
- Durasi Batch : Durasi setiap batch.
- Durasi Operasi : waktu yang dibutuhkan untuk melakukan berbagai operasi dalam milidetik.
Transaksi yang dipantau tercantum di bawah ini:
addBatch: waktu yang dihabiskan untuk membaca data masukan kelompok mikro dari sumber, memprosesnya, dan menulis data keluaran kelompok untuk disinkronkan. Ini biasanya menghabiskan sebagian besar waktu batch mikro.getBatch: waktu yang dibutuhkan untuk menyiapkan permintaan logis untuk membaca data masukan dari paket mikro saat ini dari sumber.getOffset: waktu yang dihabiskan untuk menanyakan sumber apakah mereka memiliki masukan baru.walCommit: Menulis offset ke log metadata.queryPlanning: Buat rencana eksekusi.
Perlu dicatat bahwa tidak semua operasi yang terdaftar akan ditampilkan di UI. Ada operasi yang berbeda dengan jenis sumber data yang berbeda, sehingga beberapa operasi yang terdaftar dapat dilakukan dalam satu permintaan streaming.
Memecahkan Masalah Kinerja Streaming Menggunakan UI
Di bagian ini, kita akan melihat beberapa kasus di mana Streaming Terstruktur UI baru menunjukkan bahwa sesuatu yang tidak biasa sedang terjadi. Permintaan demo tingkat tinggi terlihat seperti ini, dan dalam setiap kasus kami akan mengasumsikan beberapa prasyarat:
import java.util.UUID
val bootstrapServers = ...
val topics = ...
val checkpointLocation = "/tmp/temporary-" + UUID.randomUUID.toString
val lines = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", bootstrapServers)
.option("subscribe", topics)
.load()
.selectExpr("CAST(value AS STRING)")
.as[String]
val wordCounts = lines.flatMap(_.split(" ")).groupBy("value").count()
val query = wordCounts.writeStream
.outputMode("complete")
.format("console")
.option("checkpointLocation", checkpointLocation)
.start()Peningkatan latensi karena daya pemrosesan tidak mencukupi
Dalam kasus pertama, kami menjalankan permintaan untuk memproses data Apache Kafka sesegera mungkin. Untuk setiap batch, tugas streaming memproses semua data yang tersedia di Kafka. Jika daya pemrosesan tidak cukup untuk menangani data burst, latensi akan meningkat dengan cepat. Penilaian paling intuitif adalah bahwa Input Rows dan Batch Duration akan tumbuh secara linier. Parameter Input Rows menetapkan bahwa tugas streaming dapat menangani maksimum 8000 penulisan per detik. Tapi Tingkat Input saat ini sekitar 20.000 rekaman per detik. Kami dapat menyediakan pekerjaan penguliran dengan lebih banyak sumber daya untuk dijalankan, atau kami dapat menambahkan partisi yang cukup untuk menangani semua konsumen yang diperlukan untuk mengikuti produsen.
Stabil tetapi latensi tinggi
Bagaimana kasus ini berbeda dari yang sebelumnya? Latensi tidak meningkat, tetapi tetap stabil, seperti yang ditunjukkan pada tangkapan layar berikut:
Kami menemukan bahwa Kecepatan Proses dapat tetap stabil pada Kecepatan Input yang sama. Ini berarti bahwa kekuatan pemrosesan pekerjaan cukup untuk memproses data masukan. Namun, waktu pemrosesan untuk setiap batch, yaitu penundaan, masih 20 detik. Alasan utama latensi tinggi adalah terlalu banyak data di setiap kelompok. Kami biasanya dapat mengurangi latensi dengan meningkatkan paralelisme pekerjaan ini. Setelah menambahkan 10 lagi partisi Kafka dan 10 core untuk tugas Spark, kami menemukan latensi sekitar 5 detik - jauh lebih baik daripada 20 detik.
Gunakan Diagram Durasi Operasi untuk Pemecahan Masalah
Bagan Durasi Operasi menampilkan jumlah waktu yang dihabiskan untuk melakukan berbagai operasi dalam milidetik. Ini berguna untuk memahami waktu setiap kelompok dan mempermudah pemecahan masalah. Mari gunakan pekerjaan peningkatan kinerja " SPARK-30915 : Hindari membaca file log metadata saat mencari ID batch terbaru" di komunitas Apache Spark sebagai contoh.
Sebelum peningkatan ini, setiap batch berikutnya setelah kompresi membutuhkan waktu lebih lama daripada batch lainnya, saat log metadata yang dikompresi menjadi sangat besar.
Setelah memeriksa kode, pembacaan yang tidak perlu dari file log terkompresi ditemukan dan diperbaiki. Diagram Durasi Operasi berikut mengkonfirmasi efek yang diharapkan:
Rencana untuk masa depan
Seperti yang ditunjukkan di atas, Streaming Terstruktur UI baru akan membantu pengembang mengontrol pekerjaan streaming mereka dengan lebih baik dengan memiliki informasi yang jauh lebih berguna tentang permintaan streaming. Sebagai versi awal, UI baru masih dalam pengembangan dan akan ditingkatkan di rilis mendatang. Ada beberapa fitur yang dapat diimplementasikan dalam waktu yang tidak terlalu lama, termasuk namun tidak terbatas pada berikut ini:
- Pelajari lebih lanjut tentang menjalankan permintaan streaming: data terlambat, tanda air, metrik status data, dan banyak lagi.
- Dukungan UI Streaming Terstruktur di Spark History Server.
- Petunjuk yang lebih terlihat untuk perilaku tidak biasa: latensi, dll.
Coba UI baru
Coba UI Streaming Spark baru ini di Apache Spark 3.0 di Databricks Runtime 7.1 baru. Jika Anda menggunakan notebook Databricks, ini juga akan memberi Anda cara mudah untuk mengamati status permintaan streaming apa pun di notebook dan mengelola permintaan Anda . Anda dapat mendaftar untuk mendapatkan akun gratis di Databricks dan memulai dalam beberapa menit secara gratis, tanpa informasi kredit apa pun.
Kualitas data di DWH adalah konsistensi Data Warehouse. webinar gratis.
Bacaan yang direkomendasikan:
Alat Bangun Data, atau kesamaan Data Warehouse dan Smoothie
Menyelam ke Delta Lake: Penegakan Skema dan Evolusi
Apache Parquet Berkecepatan Tinggi dengan Python dengan Apache Arrow