Menyiapkan mode real-time

Halaman ini menjelaskan cara menjalankan kueri mode waktu nyata secara langsung dengan API Structured Streaming pada classic compute. Untuk tutorial langkah demi langkah, lihat Tutorial: Menjalankan beban kerja streaming real time. Untuk informasi konseptual tentang mode waktu nyata, lihat konsep mode waktu nyata.

Mode waktu nyata di pipeline Lakeflow mendukung pipeline yang dikelola pada komputasi klasik dan serverless, dengan kebutuhan runtime, sumber, sink, dan operator yang terpisah.

Prasyarat

Untuk menggunakan mode real time, Anda harus mengonfigurasi komputasi untuk memenuhi persyaratan berikut:

  • Gunakan komputasi klasik. Mode akses khusus dan standar didukung. Mode akses standar hanya didukung untuk Python. Konfigurasi query langsung pada halaman ini tidak didukung di notebook atau pekerjaan serverless. Untuk pemrosesan waktu nyata tanpa server, gunakan mode waktu nyata di pipeline Lakeflow.
  • Gunakan Databricks Runtime 16.4 LTS ke atas.
  • Nonaktifkan penskalakan otomatis.
  • Matikan Photon.
  • Atur spark.databricks.streaming.realTimeMode.enabled ke true.
  • Nonaktifkan instans spot untuk menghindari gangguan.

Untuk beban kerja sensitif latensi dengan UDF, Databricks merekomendasikan agar Anda menggunakan mode akses khusus. Lihat Fungsi tabel.

Untuk petunjuk tentang membuat dan mengonfigurasi komputasi klasik, lihat Referensi konfigurasi komputasi.

Streaming ke gabungan streaming

Streaming untuk mengalirkan gabungan dalam memerlukan konfigurasi tambahan untuk mode real time. Outer join stream-to-stream tidak didukung. Lihat Streaming untuk melakukan streaming gabungan.

Penting

Untuk menjalankan join antar-stream dalam mode real-time dengan beberapa stream lain pada kluster yang sama, Anda harus menggunakan Databricks Runtime 18 LTS atau yang lebih baru.

Untuk mengaktifkan streaming ke gabungan streaming untuk mode real time, atur konfigurasi Spark berikut:

Python

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

Scala

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

SQL

SET spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled = true;
SET spark.sql.streaming.join.stateFormatVersion = 4;
SET spark.sql.streaming.join.stateFormatV4.enabled = true;
SET spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion = 2;
SET spark.sql.streaming.realTimeMode.controlMessage.enabled = true;

Konfigurasi kueri

Untuk menjalankan kueri dalam mode real time, Anda harus mengaktifkan pemicu real-time. Pemicu real time hanya didukung dalam mode pembaruan.

Python

query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .outputMode("update")
        # In PySpark, the realTime trigger requires specifying the interval.
        .trigger(realTime="5 minutes")
        .start()
)

Scala

import org.apache.spark.sql.streaming.Trigger

val readStream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("subscribe", inputTopic).load()
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("topic", outputTopic)
      .option("checkpointLocation", checkpointLocation)
      .outputMode("update")
      .trigger(Trigger.RealTime())
      // Trigger.RealTime can also accept an argument specifying the batch duration.
      // For example, this code sets a batch duration of 5 minutes:
      // .trigger(Trigger.RealTime("5 minutes"))
      .start()

Pemilihan ukuran

Anda dapat menjalankan satu pekerjaan waktu nyata untuk setiap sumber daya komputasi jika sumber daya tersebut memiliki slot tugas yang cukup.

Untuk berjalan dalam mode latensi rendah, jumlah total slot tugas yang tersedia harus lebih besar dari atau sama dengan jumlah tugas di semua tahap kueri.

Contoh perhitungan slot

Jenis alur Konfigurasi Slot yang diperlukan
Stateless tahap tunggal (sumber Kafka + penampungan) maxPartitions = 8 8 slot
Stateful dua tahap (sumber Kafka + acak) maxPartitions = 8, partisi shuffle = 20 28 slot (8 + 20)
Tiga tahap (sumber Kafka + acak + partisi ulang) maxPartitions = 8, dua tahap pengacakan masing-masing 20 48 slot (8 + 20 + 20)

Jika Anda tidak mengatur maxPartitions, gunakan jumlah partisi dalam topik Kafka.

Sumber daya tambahan

  • Terakhir diperbarui pada