Configurar o modo em tempo real

Esta página descreve os pré-requisitos e configurações necessários para executar consultas em modo em tempo real no Structured Streaming. Para um tutorial passo a passo, veja Tutorial: Executar uma carga de trabalho de streaming em tempo real. Para informações conceptuais sobre o modo em tempo real, veja Modo em tempo real em Streaming Estruturado.

Pré-requisitos

Para usar o modo em tempo real, deve configurar o seu cálculo para cumprir os seguintes requisitos:

• Usa computação clássica. São suportados modos de acesso dedicados e padrão. O modo de acesso padrão é suportado apenas para Python. Lakeflow Spark Declarative Pipelines e clusters serverless não são suportados.
• Use Databricks Runtime 16.4 LTS e superiores.
• Desligue o autoscaling.
• Desliga o Photon.
• Defina spark.databricks.streaming.realTimeMode.enabled como true.
• Desligue as instâncias pontuais para evitar interrupções.

Para cargas de trabalho sensíveis à latência com UDFs, o Databricks recomenda que utilize um modo de acesso dedicado. Ver funções da tabela.

Para instruções sobre como criar e configurar computação clássica, veja Referência de configuração de Computação.

Uniões entre fluxos

As junções internas entre fluxos requerem configuração adicional para o modo em tempo real. As junções externas não são suportadas. Veja Junção de fluxo para fluxo.

Para permitir as junções stream to stream em modo em tempo real, defina as seguintes configurações do Spark:

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;

Configuração da consulta

Para executar uma consulta em modo em tempo real, deve ativar o gatilho em tempo real. Os gatilhos em tempo real são suportados apenas em modo de atualização.

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

Dimensionamento computacional

Podes executar um trabalho em tempo real por recurso de computação se o cálculo tiver espaços suficientes para tarefas.

Para ser executado no modo de baixa latência, o número total de slots de tarefas disponíveis deve ser maior ou igual ao número de tarefas em todos os estágios de consulta.

Exemplos de cálculo de faixas horárias

Se não definires maxPartitions, usa o número de partições no tópico Kafka.

Recursos adicionais

• Tutorial: Executar uma carga de trabalho de streaming em tempo real
• Exemplos de modos em tempo real
• Otimizar e monitorizar o desempenho das consultas em modo em tempo real
• Referência do modo em tempo real