Catatan
Akses ke halaman ini memerlukan otorisasi. Anda dapat mencoba masuk atau mengubah direktori.
Akses ke halaman ini memerlukan otorisasi. Anda dapat mencoba mengubah direktori.
Streaming terstruktur adalah mesin pemrosesan aliran yang dapat diskalakan dan toleran terhadap kesalahan yang dibangun di Spark. Ini memperlakukan aliran data langsung sebagai tabel tempat baris baru terus ditambahkan. Streaming Terstruktur mendukung sumber file bawaan seperti CSV, JSON, ORC, dan Parquet, bersama dengan layanan olahpesan seperti Kafka dan Azure Event Hubs.
Artikel ini membahas pengaturan sumber streaming seperti Azure Event Hubs, memasukkan data streaming ke dalam tabel Delta lakehouse, mengoptimalkan kinerja penulisan dengan partisi dan batching peristiwa, serta menjalankan pekerjaan streaming secara andal di lingkungan produksi.
Menyiapkan sumber streaming
Untuk mengalirkan data ke lakehouse, pertama-tama konfigurasikan koneksi ke sumber streaming Anda. Azure Event Hubs adalah pilihan umum. Gunakan Konektor Azure Event Hubs untuk Apache Spark untuk menyambungkan aplikasi Spark Anda ke Azure Event Hubs.
Konfigurasi dasar Event Hubs memerlukan nama namespace Event Hubs, nama hub, nama kunci akses bersama, dan kelompok konsumen.
Kelompok konsumen adalah representasi dari seluruh hub acara. Grup konsumen memungkinkan beberapa aplikasi yang mengonsumsi untuk masing-masing memiliki tampilan terpisah dari aliran peristiwa dan membaca aliran tersebut secara independen dengan kecepatan dan offset mereka sendiri.
Partisi di Azure Event Hubs memungkinkan Anda memproses peristiwa dalam volume besar secara paralel. Satu prosesor memiliki kapasitas terbatas untuk menangani peristiwa per detik, sementara beberapa prosesor dapat bekerja secara paralel di seluruh partisi.
Jika terlalu banyak partisi digunakan dengan tingkat penyerapan yang rendah, pembaca partisi menangani sebagian kecil data, menyebabkan pemrosesan nonoptimal. Jumlah partisi yang ideal tergantung pada tingkat pemrosesan yang diinginkan. Saat Anda meningkatkan jumlah unit throughput di namespace, Anda mungkin memerlukan partisi tambahan agar pembaca bersamaan dapat mencapai throughput maksimum mereka.
Uji jumlah partisi terbaik untuk skenario throughput Anda. Skenario dengan throughput tinggi umumnya menggunakan 32 partisi atau lebih.
Tabel Delta sebagai tujuan streaming
Delta Lake adalah lapisan penyimpanan sumber terbuka yang menyediakan transaksi ACID (atomitas, konsistensi, isolasi, dan durabilitas) di atas penyimpanan data lake. Dalam Fabric Data Engineering, Delta Lake mendukung upsert, pemadatan data, perjalanan waktu, evolusi skema, dan penyimpanan format terbuka.
Dengan delta sebagai format output dalam writeStream, data streaming mengalir langsung ke tabel Delta. Contoh berikut membaca dari Azure Event Hubs, mengurai isi pesan, dan menulis ke tabel Delta:
import pyspark.sql.functions as f
from pyspark.sql.types import *
df = (
spark.readStream
.format("eventhubs")
.options(**ehConf)
.load()
)
Schema = StructType([
StructField("<column_name_01>", StringType(), False),
StructField("<column_name_02>", StringType(), False),
StructField("<column_name_03>", DoubleType(), True),
StructField("<column_name_04>", LongType(), True),
StructField("<column_name_05>", LongType(), True),
])
rawData = (
df
.withColumn("bodyAsString", f.col("body").cast("string"))
.select(f.from_json("bodyAsString", Schema).alias("events"))
.select("events.*")
.writeStream
.format("delta")
.option("checkpointLocation", "Files/checkpoint")
.outputMode("append")
.toTable("deltaeventstable")
)
Dalam kode, format("delta") mengatur Delta sebagai format output, outputMode("append") hanya menulis baris baru ke tabel, dan toTable("deltaeventstable") mempertahankan data yang dialirkan ke tabel Delta terkelola.
Mengoptimalkan performa streaming
Setelah penyerapan streaming dasar berfungsi, Anda dapat meningkatkan throughput dan organisasi file dengan teknik pengoptimalan di bagian berikut.
Data partisi untuk penulisan
Untuk mengoptimalkan throughput, partisi data Anda secara efektif. Pemartisian meningkatkan throughput tulis dan performa kueri hilir. Anda dapat mempartisi data dalam memori, pada disk, atau keduanya.
Pada disk — Gunakan partitionBy() untuk menata data ke dalam subdirektori berdasarkan nilai kolom. Pilih kolom dengan kardinalitas yang baik yang menghasilkan file berukuran optimal. Hindari kolom yang membuat terlalu banyak partisi kecil atau terlalu sedikit partisi besar.
Dalam memori — Gunakan repartition() atau coalesce() untuk mendistribusikan data ke seluruh simpul pekerja sebelum penulisan:
-
repartition()meningkatkan atau mengurangi partisi dengan penyusunan ulang penuh, menyeimbangkan data secara merata. -
coalesce()hanya mengurangi partisi, meminimalkan pergerakan data.
Menggabungkan kedua pendekatan berfungsi dengan baik untuk skenario throughput tinggi. Contoh berikut membagi data menjadi 48 partisi dalam memori (mencocokkan inti CPU yang tersedia) lalu mempartisi pada disk dengan dua kolom:
rawData = (
df
.withColumn("bodyAsString", f.col("body").cast("string"))
.select(f.from_json("bodyAsString", Schema).alias("events"))
.select("events.*")
.repartition(48)
.writeStream
.format("delta")
.option("checkpointLocation", "Files/checkpoint")
.outputMode("append")
.partitionBy("<column_name_01>", "<column_name_02>")
.toTable("deltaeventstable")
)
Gunakan Tulis yang Dioptimalkan
Sebagai alternatif untuk partisi manual, Optimized Write menggabungkan atau membagi partisi sebelum menulis, memaksimalkan throughput disk tanpa panggilan manual repartition() atau coalesce(). Aktifkan dengan konfigurasi Spark:
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", True)
Dengan Optimized Write diaktifkan, Anda dapat menghapus repartition() atau coalesce() dari kode Anda dan membiarkan Spark menangani ukuran partisi. Anda masih dapat menggunakan partitionBy() untuk organisasi tingkat disk.
Peristiwa batch terjadwal dengan pemicu
Untuk lebih mengoptimalkan performa penulisan, kelompokkan peristiwa sebelum menulisnya ke piringan. Secara default, Spark memproses setiap microbatch segera setelah yang sebelumnya selesai. Mengatur interval pemicu mengakumulasi data selama periode waktu tertentu dan menulisnya dalam operasi yang lebih sedikit dan lebih besar. Batch yang lebih besar menghasilkan file Delta yang lebih besar dan mengurangi overhead file kecil.
Contoh berikut memproses peristiwa dalam interval satu menit:
rawData = (
df
.withColumn("bodyAsString", f.col("body").cast("string"))
.select(f.from_json("bodyAsString", Schema).alias("events"))
.select("events.*")
.writeStream
.format("delta")
.option("checkpointLocation", "Files/checkpoint")
.outputMode("append")
.partitionBy("<column_name_01>", "<column_name_02>")
.trigger(processingTime="1 minute")
.toTable("deltaeventstable")
)
Analisis volume data masuk dan pilih interval pemrosesan yang menghasilkan file Parquet berukuran baik dalam tabel Delta.
Menjalankan pekerjaan streaming dalam produksi
Notebook Spark adalah alat yang efektif untuk mengembangkan dan menguji logika streaming. Namun, untuk beban kerja produksi yang perlu berjalan terus menerus, gunakan definisi kerja Spark sebagai gantinya. Definisi kerja Spark adalah tugas non-interaktif berorientasi kode yang berjalan pada kluster Spark dan memberikan ketahanan dan ketersediaan yang lebih besar.
Infrastruktur yang menjalankan pekerjaan streaming dapat mengalami masalah yang menghentikan pekerjaan, seperti kegagalan perangkat keras atau patching infrastruktur. Kebijakan coba lagi secara otomatis memulai ulang pekerjaan ketika berhenti secara tak terduga. Konfigurasikan kebijakan coba lagi pada definisi pekerjaan Spark untuk menentukan berapa kali untuk memulai ulang pekerjaan (hingga percobaan ulang tak terbatas) dan interval waktu antara percobaan ulang. Dengan kebijakan coba lagi diaktifkan, pekerjaan streaming Anda akan terus berjalan sampai Anda secara eksplisit menghentikannya.
Hub pemantauan Fabric mencakup tab Streaming Terstruktur dengan metrik termasuk Laju Input, Laju Proses, Baris Input, Durasi Batch, dan Durasi Operasi.