Azure Event Hubs
Azure Event Hubs adalah layanan penyerapan telemetri hyperscale yang mengumpulkan, mengubah, dan menyimpan jutaan kejadian. Sebagai platform streaming terdistribusi, Azure Event Hubs memberi Anda latensi yang rendah dan retensi waktu yang dapat dikonfigurasi, sehingga Anda dapat memasukkan telemetri dalam jumlah besar ke cloud dan membaca data dari beberapa aplikasi dengan menggunakan semantik publish-subscribe.
Artikel ini menjelaskan cara menggunakan Streaming Terstruktur dengan kluster Azure Event Hubs dan Azure Databricks.
Catatan
Azure Event Hubs menyediakan titik akhir yang kompatibel dengan Apache Kafka yang dapat Anda gunakan dengan konektor Kafka Streaming Terstruktur, tersedia di Databricks Runtime, untuk memproses pesan dari Azure Event Hubs. Databricks merekomendasikan penggunaan konektor Kafka Streaming Terstruktur untuk memproses pesan dari Azure Event Hubs.
Persyaratan
Untuk dukungan rilis saat ini, lihat "Rilis Terbaru" di file readme proyek Konektor Spark Azure Event Hubs.
Buat pustaka di ruang kerja Azure Databricks Anda dengan menggunakan koordinat Maven
com.microsoft.azure:azure-eventhubs-spark_2.11:2.3.17.Catatan
Konektor ini diperbarui secara berkala, dan versi yang lebih baru mungkin tersedia: kami sarankan agar Anda menarik konektor terbaru dari repositori Maven
Pasang pustaka yang dibuat ke kluster Anda.
Skema
Skema rekamannya adalah:
| Column | Jenis |
|---|---|
body |
biner |
partition |
string |
offset |
string |
sequenceNumber |
long |
enqueuedTime |
rentang waktu |
publisher |
string |
partitionKey |
string |
properties |
map[string,json] |
body selalu disediakan sebagai array byte. Gunakan cast("string") untuk secara eksplisit membatalkan serialisasi kolom body.
Mulai Cepat
Mari kita mulai dengan contoh singkat: WordCount. Buku catatan berikut adalah satu-satunya hal yang diperlukan untuk menjalankan WordCount menggunakan Streaming Terstruktur dengan Azure Event Hubs.
WordCount Azure Event Hubs dengan notebook Streaming Terstruktur
Konfigurasi
Bagian ini membahas pengaturan konfigurasi yang Anda butuhkan untuk menggunakan Azure Event Hubs.
Untuk panduan terperinci cara mengonfigurasi Streaming Terstruktur dengan Azure Event Hubs, lihat Panduan Integrasi Streaming Terstruktur dan Azure Event Hubs yang dikembangkan oleh Microsoft.
Untuk panduan terperinci tentang menggunakan Streaming Terstruktur, lihat Streaming di Azure Databricks.
String koneksi
String koneksi Azure Event Hubs diperlukan untuk tersambung ke layanan Azure Event Hubs. Anda bisa mendapatkan string koneksi instans Event Hubs Anda dari portal Azure atau dengan menggunakan ConnectionStringBuilder di pustaka.
Portal Azure
Saat Anda mendapatkan string koneksi dari portal Azure, string tersebut mungkin sudah atau belum memiliki kunci EntityPath. Pertimbangkan:
// Without an entity path
val without = "Endpoint=<endpoint>;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>"
// With an entity path
val with = "Endpoint=sb://<sample>;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>;EntityPath=<eventhub-name>"
Untuk tersambung ke EventHubs Anda, EntityPath harus ada. Jika string koneksi Anda tidak memilikinya, jangan khawatir.
Kelas ini akan mengurusnya:
import org.apache.spark.eventhubs.ConnectionStringBuilder
val connectionString = ConnectionStringBuilder(without) // defined in the previous code block
.setEventHubName("<eventhub-name>")
.build
ConnectionStringBuilder
Atau, Anda dapat menggunakan ConnectionStringBuilder untuk membuat string koneksi Anda.
import org.apache.spark.eventhubs.ConnectionStringBuilder
val connectionString = ConnectionStringBuilder()
.setNamespaceName("<namespace-name>")
.setEventHubName("<eventhub-name>")
.setSasKeyName("<key-name>")
.setSasKey("<key>")
.build
EventHubsConf
Semua konfigurasi yang berkaitan dengan Azure Event Hubs terjadi di EventHubsConf. Untuk membuat EventHubsConf, Anda harus meneruskan string koneksi:
val connectionString = "<event-hub-connection-string>"
val eventHubsConf = EventHubsConf(connectionString)
Lihat String Koneksi untuk informasi lebih lanjut mengenai cara mendapatkan string koneksi yang valid.
Untuk daftar lengkap konfigurasi, lihat EventHubsConf. Berikut adalah subset konfigurasi sebagai awalannya:
| Opsi | Value | Default | Jenis kueri | Deskripsi |
|---|---|---|---|---|
consumerGroup |
String | “$Default” | Streaming dan batch | Grup konsumen adalah tampilan seluruh hub kejadian. Grup konsumen memungkinkan beberapa aplikasi yang mengonsumsi untuk memiliki tampilan terpisah dari aliran peristiwa, dan untuk membaca aliran secara mandiri dengan kecepatan mereka sendiri dengan offset mereka sendiri. Informasi lebih lanjut tersedia di dokumentasi Microsoft. |
startingPosition |
EventPosition | Mulai aliran | Streaming dan batch | Posisi awal untuk pekerjaan Streaming Terstruktur Anda. Lihat startingPositions untuk informasi mengenai urutan pembacaan opsi. |
maxEventsPerTrigger |
long | partitionCount 1000* |
Kueri streaming | Batas tarif pada jumlah maksimum kejadian yang diproses untuk setiap interval pemicu. Jumlah total kejadian yang ditentukan akan dibagi secara proporsional di seluruh partisi volume yang berbeda. |
Untuk setiap opsi, ada pengaturan yang sesuai di EventHubsConf. Misalnya:
import org.apache.spark.eventhubs.
val cs = "<your-connection-string>"
val eventHubsConf = EventHubsConf(cs)
.setConsumerGroup("sample-cg")
.setMaxEventsPerTrigger(10000)
EventPosition
EventHubsConf mengizinkan pengguna untuk menentukan posisi awal (dan akhir) dengan kelas EventPosition. EventPosition menentukan posisi kejadian dalam partisi Event Hub. Posisi dapat berupa waktu antrean, perbedaan waktu, nomor urut, awal aliran, atau akhir aliran.
import org.apache.spark.eventhubs._
EventPosition.fromOffset("246812") // Specifies offset 246812
EventPosition.fromSequenceNumber(100L) // Specifies sequence number 100
EventPosition.fromEnqueuedTime(Instant.now) // Any event after the current time
EventPosition.fromStartOfStream // Specifies from start of stream
EventPosition.fromEndOfStream // Specifies from end of stream
Jika Anda ingin memulai (atau mengakhiri) pada posisi tertentu, cukup buat EventPosition yang benar dan atur di EventHubsConf Anda:
val connectionString = "<event-hub-connection-string>"
val eventHubsConf = EventHubsConf(connectionString)
.setStartingPosition(EventPosition.fromEndOfStream)
Streaming Terstruktur Produksi dengan Azure Event Hubs
Ketika Anda menjalankan kueri streaming dalam produksi, Anda mungkin menginginkan lebih banyak jaminan ketahanan dan waktu aktif daripada yang Anda miliki ketika Anda hanya melampirkan buku catatan ke kluster dan menjalankan kueri streaming Anda secara interaktif. Impor dan jalankan buku catatan berikut yang berisi demonstrasi cara mengonfigurasi dan menjalankan Streaming Terstruktur dalam produksi dengan Azure Event Hubs dan Azure Databricks.
Untuk informasi selengkapnya, lihat Pertimbangan produksi untuk Streaming Terstruktur.