Realtimemodus instellen

Deze pagina beschrijft hoe u real-time mode queries direct kunt uitvoeren met de Structured Streaming API's op klassieke compute. Zie Zelfstudie: Een realtime streamingworkload uitvoeren voor een stapsgewijze zelfstudie. Voor conceptuele informatie over real-time modus, zie real-time mode concepts.

Realtime-modus in Lakeflow-pijplijnen ondersteunt beheerde pijplijnen op klassieke en serverloze compute, met aparte runtime-, bron-, sink- en operatorvereisten.

Voorwaarden

Als u de realtimemodus wilt gebruiken, moet u uw rekenproces configureren om te voldoen aan de volgende vereisten:

• Gebruik klassieke berekeningen. Specifieke en standaardtoegangsmodi worden ondersteund. De standaardtoegangsmodus wordt alleen ondersteund voor Python. De directe queryconfiguratie op deze pagina wordt niet ondersteund in serverless notebooks of taken. Voor serverless realtime verwerking, gebruik de realtime-modus in Lakeflow-pipelines.
• Gebruik Databricks Runtime 16.4 LTS en hoger.
• Schakel automatisch schalen uit.
• Schakel Photon uit.
• Stel spark.databricks.streaming.realTimeMode.enabled in op true.
• Schakel spot-exemplaren uit om onderbrekingen te voorkomen.

Voor latentiegevoelige workloads met UDF's raadt Databricks u aan om de toegewezen toegangsmodus te gebruiken. Zie tabelfuncties.

Zie de referentie voor compute-configuratie voor instructies over het maken en configureren van klassieke berekeningen.

Streamen naar stream-joins

Stream-naar-stream-inner joins vereisen extra configuratie voor de real-timemodus. Outer joins worden niet ondersteund. Zie stream-naar-stream-join.

Als u stream-joins wilt inschakelen voor de realtime-modus, stelt u de volgende Spark-configuraties in:

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;

Queryconfiguratie

Als u een query in realtime wilt uitvoeren, moet u de realtime-trigger inschakelen. Realtime-triggers worden alleen ondersteund in de updatemodus.

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

Berekening van de computergrootte

U kunt één realtime taak per computing resource uitvoeren als er voldoende taaksloten beschikbaar zijn.

Als u in lage-latentiemodus wilt werken, moet het totale aantal beschikbare taakslots groter zijn dan of gelijk zijn aan het aantal taken over alle querystadia heen.

Voorbeelden van sleufberekeningen

Als u maxPartitions niet instelt, gebruikt u het aantal partities van het Kafka-topic.

Aanvullende bronnen

• Zelfstudie: Een realtime streamingworkload uitvoeren
• Voorbeelden van realtimemodus
• Prestaties van realtime-modusquery's optimaliseren en bewaken
• Verwijzing naar realtimemodus