Konfigurera realtidsläge

Den här sidan beskriver hur man kör realtidsbaserade frågor direkt med Structured Streaming API:erna på klassisk compute. För en steg-för-steg-självstudie, se Självstudie: Kör en arbetsbelastning för realtidsströmning. För konceptuell information om realtidsläge, se begrepp i realtidsläge.

Realtidsläge i Lakeflow-pipelines har stöd för hanterade pipelines på klassisk och serverless compute, med separata krav på körtid, källa, mål och operatör.

Förutsättningar

Om du vill använda realtidsläge måste du konfigurera beräkningen så att den uppfyller följande krav:

  • Använd klassisk beräkning. Dedikerade och standardåtkomstlägen stöds. Standardåtkomstläge stöds endast för Python. Direktsökningskonfigurationen på denna sida stöds inte i serverlösa notebookar eller jobb. För serverlös realtidsbearbetning, använd realtidsläge i Lakeflow-pipelines.
  • Använd Databricks Runtime 16.4 LTS och senare.
  • Inaktivera autoskalning.
  • Stäng av Foton.
  • Ställ in spark.databricks.streaming.realTimeMode.enabled på true.
  • Inaktivera spotinstanser för att undvika avbrott.

För svarstidskänsliga arbetsbelastningar med UDF:er rekommenderar Databricks att du använder dedikerat åtkomstläge. Se Tabellfunktioner.

Anvisningar om hur du skapar och konfigurerar klassisk beräkning finns i Referens för beräkningskonfiguration.

Ström-till-ström-joiner

Ström-till-ström-inre sammanslagningar kräver ytterligare konfiguration för realtidsläge. Yttre joiner mellan strömmar stöds inte. Se ström-till-ström-sammanfogning.

Important

Om du vill köra en stream-till-stream-join i realtidsläge med flera andra strömmar i samma kluster, måste du använda Databricks Runtime 18 LTS eller senare.

I Databricks Runtime 18.2 och nedan stöder Structured Streaming inte följande konfigurationer för andra bearbetningslägen, inklusive processingTime och availableNow.

Ställ in följande Spark-konfigurationer för att aktivera stream to stream-kopplingar för realtidsläge:

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;

Frågekonfiguration

Om du vill köra en fråga i realtidsläge måste du aktivera realtidsutlösaren. Realtidsutlösare stöds endast i uppdateringsläge.

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()

Beräkningsstorlek

Du kan köra ett realtidsjobb per beräkningsresurs om resursen har tillräckligt med uppgiftsplatser.

Om du vill köra i läge med låg latens måste det totala antalet tillgängliga aktivitetsfack vara större än eller lika med antalet aktiviteter i alla frågesteg.

Exempel på platsberäkningar

Typ av pipeline Konfiguration Nödvändiga fack
Tillståndslös i en fas (Kafka-källa + mottagare) maxPartitions = 8 8 platser
Tillståndskänsligt i två steg (Kafka-källa + shuffle) maxPartitions = 8, shuffle partitioner = 20 28 platser (8 + 20)
Tresteg (Kafka-källa + shuffle + ompartition) maxPartitions = 8, två blandningssteg på 20 vardera 48 platser (8 + 20 + 20)

Om du inte anger maxPartitions, använd antalet partitioner i Kafka-ämnet.

Ytterligare resurser

  • Uppdaterades senast