Configurar o modo em tempo real

Esta página descreve como executar consultas no modo em tempo real diretamente com as APIs de Streaming Estruturado na computação clássica. Para obter um tutorial passo a passo, consulte Tutorial: Executar uma carga de trabalho de streaming em tempo real. Para informações conceituais sobre o modo em tempo real, veja Conceitos do modo em tempo real.

O modo em tempo real nos pipelines Lakeflow suporta pipelines gerenciados em computação clássica e serverless, com requisitos separados de runtime, source, sink e operador.

Pré-requisitos

Para usar o modo em tempo real, você deve configurar sua computação para atender aos seguintes requisitos:

  • Use a computação clássica. Há suporte para modos de acesso dedicados e padrão. O modo de acesso padrão tem suporte apenas para Python. A configuração de consulta direta nesta página não é suportada em notebooks serverless ou jobs. Para processamento em tempo real sem servidor, use o modo em tempo real nos pipelines Lakeflow.
  • Utilize o Databricks Runtime 16.4 LTS ou versões posteriores.
  • Desativar o dimensionamento automático.
  • Desative o Photon.
  • Defina spark.databricks.streaming.realTimeMode.enabled como true.
  • Desative instâncias spot para evitar interrupções.

Para cargas de trabalho sensíveis à latência com UDFs, o Databricks recomenda que você use o modo de acesso dedicado. Consulte funções de tabela.

Para obter instruções sobre como criar e configurar a computação clássica, consulte a referência de configuração de computação.

Junções de transmissão a transmissão

Junções internas entre fluxos exigem configuração adicional no modo em tempo real. Não há suporte para junções externas. Consulte Junção de transmissão a transmissão.

Importante

Para executar uma junção entre fluxos em modo de tempo real com vários outros fluxos no mesmo cluster, você deve usar o Databricks Runtime 18 LTS ou superior.

No Databricks Runtime 18.2 e abaixo, o Streaming Estruturado não dá suporte às seguintes configurações para outros modos de processamento, incluindo processingTime e availableNow.

Para habilitar junções entre streams no 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 de consulta

Para executar uma consulta no modo em tempo real, você deve habilitar o gatilho em tempo real. Os gatilhos em tempo real têm suporte apenas no 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 de computação

Você pode executar um trabalho em tempo real por recurso de computação se a computação tiver slots de tarefa suficientes.

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

Exemplos de cálculo de slots

Tipo de pipeline Configuração Slots necessários
Sem estado de estágio único (fonte Kafka + coletor) maxPartitions = 8 8 espaços
Com estado em dois estágios (fonte Kafka + embaralhamento) maxPartitions = 8, partições de embaralhamento = 20 28 slots (8 + 20)
Três estágios (fonte Kafka + embaralhamento + repartição) maxPartitions = 8, duas fases de shuffle de 20 cada um 48 vagas (8 + 20 + 20)

Se você não definir maxPartitions, use o número de partições no tópico Kafka.

Recursos adicionais

  • Última atualização em