Valós idejű mód beállítása

Ez a lap a strukturált streamelésben való valós idejű módú lekérdezések futtatásához szükséges előfeltételeket és konfigurációt ismerteti. Részletes oktatóanyagért tekintse meg az oktatóanyagot: Valós idejű streamelési számítási feladat futtatása. A valós idejű módról szóló fogalmi információkért lásd: Valós idejű mód fogalmak.

Előfeltételek

A valós idejű mód használatához úgy kell konfigurálnia a számítást, hogy megfeleljen a következő követelményeknek:

  • Használjon klasszikus számítást. A dedikált és standard hozzáférési módok támogatottak. A standard hozzáférési módot csak a Python támogatja. A Lakeflow-folyamatok és a kiszolgáló nélküli fürtök nem támogatottak.
  • Használja a Databricks Runtime 16.4 LTS-t és újabb verziót.
  • Kapcsolja ki az automatikus skálázást.
  • Kapcsolja ki a Photont.
  • Állítsa a spark.databricks.streaming.realTimeMode.enabled elemet true értékre.
  • Kapcsolja ki a spot példányokat a megszakítások elkerülése érdekében.

Az UDF-ekkel rendelkező késésre érzékeny számítási feladatok esetében a Databricks azt javasolja, hogy dedikált hozzáférési módot használjon. Lásd Táblafüggvények.

A klasszikus számítás létrehozásával és konfigurálásával kapcsolatos utasításokért tekintse meg a számítási konfigurációs referenciát.

Adatfolyamok közötti összekapcsolások

Az adatfolyamok közötti belső összekapcsolások további konfigurációt igényelnek valós idejű üzemmódban. A külső összekapcsolások nem támogatottak. Lásd: Adatfolyamok összekapcsolása.

Important

Ha valós idejű módban szeretne stream–stream összekapcsolást futtatni ugyanazon a fürtön több másik streammel, a Databricks Runtime 18 LTS vagy újabb verzióját kell használnia.

A Databricks Runtime 18.2-s és újabb verziójában a strukturált streamelés nem támogatja a következő konfigurációkat más feldolgozási módokhoz, beleértve processingTime és availableNow.

A valós idejű módban történő stream–stream összekapcsolások engedélyezéséhez állítsa be a következő Spark-konfigurációkat:

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;

Lekérdezéskonfiguráció

Ha valós idejű módban szeretne lekérdezést futtatni, engedélyeznie kell a valós idejű eseményindítót. A valós idejű eseményindítók csak frissítési módban támogatottak.

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.execution.streaming.RealTimeTrigger

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(RealTimeTrigger.apply())
      // RealTimeTrigger can also accept an argument specifying the checkpoint interval.
      // For example, this code indicates a checkpoint interval of 5 minutes:
      // .trigger(RealTimeTrigger.apply("5 minutes"))
      .start()

Számítási méretezés

Számítási erőforrásonként egy valós idejű feladatot futtathat, ha a számítás elegendő feladathellyel rendelkezik.

Alacsony késésű módban való futtatáshoz a rendelkezésre álló feladathelyek teljes számának meg kell haladnia vagy egyenlőnek kell lennie az összes lekérdezési fázisban lévő tevékenységek számával.

Pontszámítási példák

Folyamat típusa Konfiguráció Szükséges tárolóhelyek
Egyfázisú állapot nélküli (Kafka forrás + fogadó) maxPartitions = 8 8 tárolóhely
Kétfázisú állapotalapú (Kafka forrás + shuffle) maxPartitions = 8, keverési partíciók = 20 28 hely (8 + 20)
Háromfázisú (Kafka forrás + shuffle + újrapartíció) maxPartitions = 8, két 20 lépéses keverési szakasz 48 helyek (8 + 20 + 20)

Ha nem állítja be maxPartitions, használja a Kafka téma partícióinak számát.

További erőforrások

  • Utolsó frissítés időpontja: