Configurar modo em tempo real

Esta página descreve os pré-requisitos e a configuração necessários para executar queries do modo em tempo real no Structured Streaming. Para um tutorial passo a passo, consulte Tutorial: Executar uma carga de trabalho de transmissão em tempo real. Para obter informações conceituais sobre o modo em tempo real, consulte Conceitos do modo em tempo real.

Pré-requisitos

Para usar o modo tempo real, você deve configurar seu compute para atender aos seguintes requisitos:

Use o compute clássico. Os modos de acesso dedicado e padrão são compatíveis. O modo de acesso padrão é compatível apenas com Python. LakeFlow Pipelines e clusters serverless não são compatíveis.
Utilize Databricks Runtime 16.4 LTS ou superior.
Desative a escala automática.
Desligue o Photon.
Defina spark.databricks.streaming.realTimeMode.enabled como true.
Desative as instâncias spot para evitar interrupções.

Para cargas de trabalho sensíveis à latência com UDFs (Funções Definidas pelo Usuário), a Databricks recomenda o uso do modo de acesso dedicado. Consulte as funções da tabela.

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

joins de transmissão para transmissão

Joins internos entre transmissões exigem configuração adicional para o modo de tempo real. Outer joins não são compatíveis. Consulte join entre transmissões.

importante

Para executar um join Stream a Stream em modo de tempo real com vários outros Stream no mesmo cluster, é necessário usar o Databricks Runtime 18 LTS e acima.

No Databricks Runtime 18.2 e abaixo, a transmissão estructurada não oferece suporte para as seguintes configurações para outros modos de processamento, incluindo processingTime e availableNow.

Para habilitar joins de transmissão para transmissão para o modo de tempo real, defina as seguintes configurações do Spark:

Python
Scala
SQL

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 no modo tempo real, é necessário habilitar o trigger tempo real. Os gatilhos em tempo real são suportados apenas no modo de atualização.

Python
Scala

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 cálculo

Você pode executar um Job de tempo real por recurso compute se a compute tiver slots de tarefa suficientes.

Para execução em 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 da consulta.

Exemplos de cálculo de slots

tipo de tubulação	Configuração	Vagas necessárias
Sem estado em estágio único (fonte + destino Kafka)	`maxPartitions` = 8	8 vagas
Com estado em dois estágios (fonte Kafka + embaralhamento)	`maxPartitions` = 8, embaralhar partições = 20	28 vagas (8 + 20)
Três estágios (fonte Kafka + shuffle + repartição)	`maxPartitions` = 8, duas fases de embaralhamento de 20 cada	48 vagas (8 + 20 + 20)

tipo de tubulação	Configuração	Vagas necessárias
Sem estado em estágio único (fonte + destino Kafka)	`maxPartitions` = 8	8 vagas
Com estado em dois estágios (fonte Kafka + embaralhamento)	`maxPartitions` = 8, embaralhar partições = 20	28 vagas (8 + 20)
Três estágios (fonte Kafka + shuffle + repartição)	`maxPartitions` = 8, duas fases de embaralhamento de 20 cada	48 vagas (8 + 20 + 20)

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

Pré-requisitos​

joins de transmissão para transmissão​

Configuração da consulta​

dimensionamento de cálculo​

Exemplos de cálculo de slots​

Recursos adicionais​

Pré-requisitos

joins de transmissão para transmissão

Configuração da consulta

dimensionamento de cálculo

Exemplos de cálculo de slots

Recursos adicionais