Pular para o conteúdo principal

modo tempo real em transmissão estruturada

info

Visualização

Esse recurso está em Public Preview.

Esta página descreve o modo tempo real, um tipo de gatilho para transmissão estruturada que permite o processamento de dados com latência ultrabaixa, com latência de ponta a ponta de apenas 5 ms. Este modo foi projetado para cargas de trabalho operacionais que exigem resposta imediata aos dados de transmissão.

LTS O modo tempo real está disponível no Databricks Runtime, 16.4 e versões superiores.

Cargas de trabalho operacionais​

As cargas de trabalho de transmissão podem ser amplamente divididas em cargas de trabalho analíticas e cargas de trabalho operacionais:

  • As cargas de trabalho analíticas utilizam a ingestão de dados e transformações, seguindo normalmente a arquitetura medalhão (por exemplo, ingestão de dados nas tabelas bronze, prata e ouro).
  • As cargas de trabalho operacionais consomem dados em tempo real, aplicam lógica de negócios e acionam ações ou decisões posteriores.

Alguns exemplos de cargas de trabalho operacionais são:

  • Bloquear ou sinalizar uma transação com cartão de crédito em tempo real se uma pontuação de fraude exceder um limite, com base em fatores como localização incomum, grande valor da transação ou padrões de gastos rápidos.
  • Entregar uma mensagem promocional quando os dados do clickstream mostram que um usuário está procurando jeans há cinco minutos, oferecendo um desconto de 25% se ele comprar nos próximos 15 minutos.

Em geral, as cargas de trabalho operacionais são caracterizadas pela necessidade de latência de ponta a ponta de menos de um segundo. Isso pode ser alcançado com o modo tempo real em uma transmissão estruturada Apache Spark.

Como o modo em tempo real alcança baixa latência​

O modo em tempo real aprimora a arquitetura de execução ao:

  • Executar lotes de longa duração (o “ default ” é de 5 minutos), nos quais os dados são processados à medida que ficam disponíveis na fonte.
  • Todos os estágios da consulta são agendados simultaneamente. Isso requer que o número de slots de tarefa disponíveis seja igual ou superior ao número de tarefas de todas as etapas em um lote.
  • Os dados são transmitidos entre as etapas assim que são produzidos, utilizando um shuffle de transmissão.

Ao final do processamento de um lote, e antes do próximo lote começar, a transmissão estruturada verifica o progresso dos pontos de controle e disponibiliza as métricas do último lote. Se os lotes forem mais longos, essas atividades podem ser menos frequentes, levando a repetições mais longas em caso de falha e atraso na disponibilidade das métricas. Por outro lado, se os lotes forem menores, essas atividades se tornam mais frequentes, podendo afetar a latência. A Databricks recomenda que você compare o modo em tempo real com sua carga de trabalho e seus requisitos para encontrar o intervalo de acionamento adequado.

configuração de agrupamento​

Para utilizar o modo tempo real na transmissão estruturada, é necessário configurar um LakeFlow Job clássico:

  1. No seu Databricks workspace, clique em Novo no canto superior esquerdo. Selecione Mais e clique em agrupamento .

  2. Limpe a aceleração doPhoton .

  3. Desative a opção “Ativar escala automática ”.

  4. Em Desempenho avançado , desmarque a opção Usar instâncias spot .

  5. Em Modo Avançado e de Acesso, clique em Manual e selecione Dedicado (anteriormente: Usuário único).

  6. Em Spark , insira o seguinte em Spark config :

    spark.databricks.streaming.realTimeMode.enabled true

  7. Clique em Criar .

requisitos de tamanho do agrupamento​

É possível executar uma tarefa em tempo real por agrupamento, caso o agrupamento possua slots de tarefas suficientes.

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

Exemplos de cálculo de slots​

pipeline e sem estado de estágio único (Kafka e fonte + coletor):

Se maxPartitions = 8, você precisará de pelo menos 8 slots. Se maxPartitions não estiver definido, utilize o número de partições do tópico Kafka.

pipeline e com estado em duas etapas (Kafka e fonte + shuffle):

Se maxPartitions = 8 e shuffle partitions = 20, você precisará de 8 + 20 = 28 slots.

pipeline e em três etapas (Kafka fonte + shuffle + repartição):

Com maxPartitions = 8 e dois estágios aleatórios de 20 cada, você precisa de 8 + 20 + 20 = 48 slots.

considerações importantes​

Ao configurar o clustering, considere o seguinte:

  • Ao contrário do modo micro-lotes, a tarefa em tempo real pode permanecer parada enquanto aguarda dados, portanto, o dimensionamento correto é essencial para evitar desperdício de recursos.

  • Almeje um nível de utilização alvo (por exemplo, 50%) ajustando:

    • maxPartitions (para Kafka)
    • spark.sql.shuffle.partitions (para fases aleatórias)

  • A Databricks recomenda definir maxPartitions para que cada tarefa lide com várias partições do Kafka, a fim de reduzir a sobrecarga.

  • Ajuste os slots de tarefas por worker para corresponder à carga de trabalho para tarefas simples de uma etapa.

  • Para tarefas com muitas reorganizações, experimente encontrar o número mínimo de partições de reorganização que evitem atrasos e ajuste a partir daí. A tarefa não será agendada se o agrupamento não tiver slots suficientes.

nota

A partir de Databricks Runtime 16.4 LTS e acima, todos os pipelines tempo real utilizam o checkpoint v2, que permite a alternância perfeita entre os modos tempo real e micro-lotes.

Configuração de consulta​

É necessário habilitar o gatilho em tempo real para especificar que uma consulta deve ser executada usando o modo de baixa latência. Além disso, os acionadores em tempo real são suportados apenas no modo de atualização. Por exemplo:

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, realTime trigger requires you to specify the interval.
        .trigger(realTime="5 minutes")
        .start()
)

Observabilidade​

Anteriormente, a latência de consulta de ponta a ponta estava intimamente ligada à duração dos lotes, tornando a duração dos lotes um bom indicador da latência da consulta. No entanto, este método não se aplica mais no modo em tempo real, exigindo abordagens alternativas para medir a latência. A latência de ponta a ponta é específica da carga de trabalho e, às vezes, só pode ser medida com precisão com a lógica comercial. Por exemplo, se o carimbo de data/hora da fonte for emitido no Kafka, a latência pode ser calculada como a diferença entre o carimbo de data/hora de saída do Kafka e o carimbo de data/hora da fonte.

É possível estimar a latência de ponta a ponta de várias maneiras, com base em informações parciais coletadas durante o processo de transmissão.

Use o StreamingQueryProgress​

As seguintes métricas estão incluídas no evento “ StreamingQueryProgress ”, que é automaticamente registrado no driver “ logs”. Você também pode acessá-los por meio da função de retorno de chamada onQueryProgress() do StreamingQueryListener. QueryProgressEvent.json() ou toString() incluem métricas adicionais do modo em tempo real.

  1. Latência de processamento (ProcessingLatencyMS ). O tempo decorrido entre o momento em que a consulta no modo em tempo real lê um registro e antes de ele ser gravado no próximo estágio ou downstream. Para consultas de estágio único, isso mede a mesma duração que a latência E2E. Essas métricas são relatadas por tarefa.
  2. Latência do enfileiramento de origem (SourceQueuingLatencyMS) . O tempo decorrido entre o momento em que um registro é gravado com sucesso em um barramento de mensagens, por exemplo, o tempo de acréscimo ao log no Kafka, e o momento em que o registro foi lido pela primeira vez pela consulta em modo em tempo real. Essas métricas são relatadas por tarefa.
  3. Latência E2E (latências E2E) . O tempo entre o momento em que o registro é gravado com sucesso em um barramento de mensagens e o momento em que o registro é gravado a jusante pela consulta em modo tempo real. Essas métricas são agregadas por lotes em todos os registros processados por todas as tarefas.

Por exemplo:

"rtmMetrics" : {
    "processingLatencyMs" : {
      "P0" : 0,
      "P50" : 0,
      "P90" : 0,
      "P95" : 0,
      "P99" : 0
    },
    "sourceQueuingLatencyMs" : {
      "P0" : 0,
      "P50" : 1,
      "P90" : 1,
      "P95" : 2,
      "P99" : 3
    },
    "e2eLatencyMs" : {
      "P0" : 0,
      "P50" : 1,
      "P90" : 1,
      "P95" : 2,
      "P99" : 4
    },

Utilizar “ API ” na função​

O recurso "Observar" API auxilia na medição da latência sem iniciar outra tarefa. Se você tiver um carimbo de data/hora da fonte que se aproxime da hora de chegada dos dados da fonte e ele for passado antes de chegar ao destino, ou se você puder encontrar uma maneira de passar o carimbo de data/hora, você pode estimar a latência de cada lote usando o Observe API:

Python
from datetime import datetime

from pyspark.sql.functions import avg, col, lit, max, percentile_approx, udf, unix_millis
from pyspark.sql.types import TimestampType

@udf(returnType=TimestampType())
def current_timestamp():
  return datetime.now()

# Query before outputting
.withColumn("temp-timestamp", current_timestamp())
.withColumn(
  "latency",
  unix_millis(col("temp-timestamp")).cast("long") - unix_millis(col("timestamp")).cast("long"))
.observe(
  "observedLatency",
  avg(col("latency")).alias("avg"),
  max(col("latency")).alias("max"),
  percentile_approx(col("latency"), lit(0.99), lit(150)).alias("p99"),
  percentile_approx(col("latency"), lit(0.5), lit(150)).alias("p50"))
.drop(col("latency"))
.drop(col("temp-timestamp"))
# Output part of the query. For example, .WriteStream, etc.

Neste exemplo, um timestamp atual é registrado antes da saída da entrada, e a latência é estimada calculando a diferença entre esse timestamp e o timestamp de origem do registro. Os resultados são incluídos nos relatórios de progresso e disponibilizados aos ouvintes. Aqui está um exemplo de saída:

"observedMetrics" : {
  "observedLatency" : {
    "avg" : 63.8369765176552,
    "max" : 219,
    "p99" : 154,
    "p50" : 49
  }
}

O que é suportado?​

Ambientes​

Tipo de cluster

Suportado

Dedicado (anteriormente: usuário único)

Sim

Padrão (anteriormente: compartilhado)

Não

LakeFlow Pipeline declarativo clássico

Não

LakeFlow Servidor de pipeline declarativo sem servidor

Não

Serverless

Não

Idiomas​

Idioma

Suportado

Scala

Sim

Java

Sim

Python

Sim

Modos de execução​

Execução Mode

Suportado

Modo de atualização

Sim

Modo de anexação

Não

Modo completo

Não

Fontes​

Fontes

Suportado

Apache Kafka

Sim

AWS MSK

Sim

Eventhub (utilizando o Conector Kafka)

Sim

Kinesis

Sim (somente modo EFO)

Google Pub/Sub

Não

Apache Pulsar

Não

Pias​

Pias

Suportado

Apache Kafka

Sim

Eventhub (utilizando o Conector Kafka)

Sim

Kinesis

Não

Google Pub/Sub

Não

Apache Pulsar

Não

Dissipadores arbitrários (usando ForEachWriter)

Sim

Operadores​

Operadores

Suportado

Operações sem estado

  • Seleção

Sim

  • Projeção

Sim

UDFs

  • UDF Scala

Sim

  • UDF Python

Sim (com algumas limitações)

Agregação

  • soma

Sim

  • contar

Sim

  • máximo

Sim

  • min

Sim

  • média

Sim

Funções de agregação

Sim

Janelamento

  • Caindo

Sim

  • Deslizando

Sim

  • Sessão

Não

Desduplicação

  • Elimine duplicatas

Sim (o estado é ilimitado)

  • Elimine duplicatas dentro da marca d'água

Não

transmissão - Junção de tabelas

  • Tabela de transmissão (deve ser pequena)

Sim

transmissão - transmissão join

Não

Grupos de mapas (planos) com estado

Não

Transformar com estado

Sim (com algumas diferenças)

união

Sim (com algumas limitações)

Para cada

Sim

Para cada lote

Não

Partições do mapa

Sim

Utilize transformWithState no modo em tempo real.​

Para a criação de aplicativos personalizados com estado, o Databricks suporta transformWithState, um API em Apache Spark transmissão estruturada. Consulte Criar um aplicativo com estado personalizado para obter mais informações sobre o API e trechos de código.

No entanto, existem algumas diferenças entre o comportamento do API no modo tempo real e as consultas de transmissão tradicionais que utilizam a arquitetura de micro-lotes.

  • O método no modo em tempo real handleInputRows(key: String, inputRows: Iterator[T], timerValues: TimerValues) é chamado para cada linha.

    • O iterador inputRows retorna um único valor. No modo micro lotes, ele é chamado uma vez para cada key, e o iterador inputRows retorna todos os valores para um key nos micro lotes.
    • Você deve estar ciente dessa diferença ao escrever seu código.

  • Os temporizadores de eventos não são suportados no modo em tempo real.

  • No modo em tempo real, os temporizadores são atrasados em sua ativação, dependendo da chegada dos dados. Caso contrário, se não houver dados, ele será disparado no final dos lotes de longa execução. Por exemplo, se um cronômetro deve disparar às 10:00:00 e não houver chegada de dados simultaneamente, ele não será acionado. Em vez disso, se os dados chegarem às 10:00:10, o cronômetro será acionado com um atraso de 10 segundos. Ou, se nenhum dado chegar e os lotes de longa execução estiverem sendo encerrados, execute o temporizador antes de encerrar os lotes de longa execução.

UDFs do Python​

O Databricks suporta a maioria das funções definidas pelo usuário (UDFs) do Python no modo em tempo real:

Tipo de UDF

Suportado

UDF sem estado

  • UDF escalar Python(link)

Sim

  • UDF escalar de seta

Sim

  • UDF escalar do Pandas(link)

Sim

  • Função de seta (mapInArrow)

Sim

Sim

UDF de agrupamento com estado (UDAF)

  • transformWithState (NOTA: apenas a interface Row)

Sim

  • Inscreva-se em Pandas com o estado

Não

UDF de agrupamento não fatídico (UDAF)

  • aplicar

Não

  • Aplicar na seta

Não

  • Aplique em Pandas

Não

Função de tabela

- UDTF (link)

Não

UC UDF

Não

Há vários pontos a serem considerados ao usar UDFs do Python no modo de tempo real:

  • Para minimizar a latência, configure o tamanho dos lotes do Arrow (spark.sql.execution.arrow.maxRecordsPerBatch) como 1.

    • Compensação: essa configuração otimiza a latência em detrimento da taxa de transferência. Para a maioria das cargas de trabalho, essa configuração é recomendada.
    • Aumente o tamanho dos lotes somente se for necessária uma taxa de transferência maior para acomodar o volume de entrada, aceitando o possível aumento na latência.

  • Pandas Os UDFs e as funções não funcionam bem com um tamanho de lote Arrow de 1.

    • Se o senhor usar Pandas UDFs ou funções, defina o tamanho dos lotes Arrow para um valor mais alto (por exemplo, 100 ou mais).
    • Observe que isso implica maior latência. A Databricks recomenda o uso do Arrow UDF ou da função, se possível.

  • Devido ao problema de desempenho com Pandas, transformWithState só é compatível com a interface Row.

Limitações​

Limitação da fonte​

Para o Kinesis, o modo de pesquisa não é compatível. Além disso, repartições frequentes podem afetar negativamente a latência.

Limitação sindical​

Para o Union, existem algumas limitações:

  • A autounião não é suportada:

    • Kafka : Não é possível utilizar o mesmo objeto de estrutura de dados de origem e estruturas de dados derivadas da união. Solução alternativa: use dataframes diferentes que sejam lidos da mesma fonte.
    • Kinesis : Não é possível unir estruturas de dados derivadas da mesma fonte Kinesis com a mesma configuração. Solução alternativa: além de utilizar Dataframes diferentes, é possível atribuir uma opção “consumerName” diferente a cada DataFrame.

  • Operadores com estado (por exemplo, aggregate, deduplicate, transformWithState) definidos antes da União não são suportados.

  • A união com fontes lotes não é suportada.

Exemplos​

Os exemplos abaixo mostram consultas que são suportadas.

Consultas sem estado​

Todas as consultas sem estado de um ou vários estágios são suportadas.

Fonte Kafka para Destino Kafka​

Neste exemplo, você lê de uma fonte Kafka e grava em um destino Kafka.

Python
query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("startingOffsets", "earliest")
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .trigger(realTime="5 minutes")
        .outputMode("update")
        .start()
)

Repartição​

Neste exemplo, você lê de uma fonte Kafka, reparticiona os dados em 20 partições e grava em um destino Kafka.

Devido a uma limitação atual da implementação, defina a configuração do Spark spark.sql.execution.sortBeforeRepartition como false antes de utilizar a repartição.

Python
# Sorting is not supported in repartition with real-time mode, so this has to be set to false to achieve low latency.
spark.conf.set("spark.sql.execution.sortBeforeRepartition", "false")

query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("subscribe", input_topic)
    .option("startingOffsets", "earliest")
    .load()
    .repartition(20)
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

transmissão-Instantâneo join (apenas transmissão)​

Neste exemplo, você lê de Kafka, join os dados com uma tabela estática e grava em um sink Kafka. Observe que apenas a junção transmissão-estática que transmite a tabela estática é suportada, o que significa que a tabela estática deve caber na memória.

Python
from pyspark.sql.functions import broadcast, expr

# We assume the static table in the path `stateic_table_location` has a column 'lookupKey'.

query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("subscribe", input_topic)
    .option("startingOffsets", "earliest")
    .load()
    .withColumn("joinKey", expr("CAST(value AS STRING)"))
    .join(
        broadcast(spark.read.format("parquet").load(static_table_location)),
        expr("joinKey = lookupKey")
    )
    .selectExpr("value AS key", "value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Fonte Kinesis para destino Kafka​

Neste exemplo, você lê de uma fonte Kinesis e grava em um destino Kafka.

Python
query = (
    spark.readStream
        .format("kinesis")
        .option("region", region_name)
        .option("awsAccessKey", aws_access_key_id)
        .option("awsSecretKey", aws_secret_access_key)
        .option("consumerMode", "efo")
        .option("consumerName", consumer_name)
        .load()
        .selectExpr("parttitionKey AS key", "CAST(data AS STRING) AS value")
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .trigger(realTime="5 minutes")
        .outputMode("update")
        .start()
)

União​

Neste exemplo, você une dois DataFrames do Kafka de dois tópicos diferentes e grava em um destino do Kafka.

Python
df1 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_1)
    .load()
)

df2 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_2)
    .load()
)

query = (
    df1.union(df2)
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Consultas com estado​

Desduplicação​

Python
query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic)
    .load()
    .dropDuplicates(["timestamp", "value"])
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Agregação​

Python
from pyspark.sql.functions import col

query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic)
    .load()
    .groupBy(col("timestamp"), col("value"))
    .count()
    .selectExpr("CAST(value AS STRING) AS key", "CAST(count AS STRING) AS value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

União com agregação​

Neste exemplo, primeiro você une dois DataFrames do Kafka de dois tópicos diferentes e, em seguida, realiza uma agregação. Por fim, você grava no sink Kafka.

Python
from pyspark.sql.functions import col

df1 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_1)
    .load()
)

df2 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_2)
    .load()
)

query = (
    df1.union(df2)
    .groupBy(col("timestamp"), col("value"))
    .count()
    .selectExpr("CAST(value AS STRING) AS key", "CAST(count AS STRING) AS value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Transformar com estado​

Scala
import org.apache.spark.sql.Encoders
import org.apache.spark.sql.execution.streaming.RealTimeTrigger
import org.apache.spark.sql.streaming.{ListState, MapState, StatefulProcessor, OutputMode, TTLConfig, TimeMode, TimerValues, ValueState}

/**
 * This processor counts the number of records it has seen for each key using state variables
 * with TTLs. It redundantly maintains this count with a value, list, and map state to put load
 * on the state variable cleanup mechanism. (In practice, only one value state is needed to maintain
 * the count for a given grouping key.)
 *
 * The input schema it expects is (String, Long) which represents a (key, source-timestamp) tuple.
 * The source-timestamp is passed through so that we can calculate end-to-end latency. The output
 * schema is (String, Long, Long), which represents a (key, count, source-timestamp) 3-tuple.
 *
 */

class RTMStatefulProcessor(ttlConfig: TTLConfig)
  extends StatefulProcessor[String, (String, Long), (String, Long, Long)] {
  @transient private var _value: ValueState[Long] = _
  @transient private var _map: MapState[Long, String] = _
  @transient private var _list: ListState[String] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    // Counts the number of records this key has seen
    _value = getHandle.getValueState("value", Encoders.scalaLong, ttlConfig)
    _map = getHandle.getMapState("map", Encoders.scalaLong, Encoders.STRING, ttlConfig)
    _list = getHandle.getListState("list", Encoders.STRING, ttlConfig)
  }

  override def handleInputRows(
      key: String,
      inputRows: Iterator[(String, Long)],
      timerValues: TimerValues): Iterator[(String, Long, Long)] = {
    inputRows.map { row =>
      val key = row._1
      val sourceTimestamp = row._2

      val oldValue = _value.get()
      _value.update(oldValue + 1)
      _map.updateValue(oldValue, key)
      _list.appendValue(key)

      (key, oldValue + 1, sourceTimestamp)
    }
  }
}

spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("subscribe", inputTopic)
      .load()
      .select(col("key").cast("STRING"), col("value").cast("STRING"), col("timestamp"))
      .as[(String, String, Timestamp)]
      .groupByKey(row => row._1)
      .transformWithState(new RTMStatefulProcessor(TTLConfig(Duration.ofSeconds(30))), TimeMode.ProcessingTime, OutputMode.Update)
      .as[(String, Long, Long)]
      .select(
            col("_1").as("key"),
            col("_2").as("value")
      )
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("topic", outputTopic)
      .option("checkpointLocation", checkpointLocation)
      .trigger(RealTimeTrigger.apply("5 minutes"))
      .outputMode(OutputMode.Update())
      .start()
nota

Existe uma diferença entre a forma como o modo tempo real e outros modos de execução na transmissão estruturada executam o StatefulProcessor em transformWithState. Consulte Utilizar transformWithState no modo em tempo real.

TransformWithState (PySpark, interface Row)​

Python
from typing import Iterator, Tuple

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import LongType, StringType, TimestampType, StructField, StructType


class RTMStatefulProcessor(StatefulProcessor):
  """
  This processor counts the number of records it has seen for each key using state variables
  with TTLs. It redundantly maintains this count with a value, list, and map state to put load
  on the state variable cleanup mechanism. (In practice, only one value state is needed to maintain
  the count for a given grouping key.)

  The input schema it expects is (String, Long) which represents a (key, source-timestamp) tuple.
  The source-timestamp is passed through so that we can calculate end-to-end latency. The output
  schema is (String, Long, Long), which represents a (key, count, source-timestamp) 3-tuple.
  """

  def init(self, handle: StatefulProcessorHandle) -> None:
    state_schema = StructType([StructField("value", LongType(), True)])
    self.value_state = handle.getValueState("value", state_schema, 30000)
    map_key_schema = StructType([StructField("key", LongType(), True)])
    map_value_schema = StructType([StructField("value", StringType(), True)])
    self.map_state = handle.getMapState("map", map_key_schema, map_value_schema, 30000)
    list_schema = StructType([StructField("value", StringType(), True)])
    self.list_state = handle.getListState("list", list_schema, 30000)

  def handleInputRows(self, key, rows, timerValues) -> Iterator[Row]:
    for row in rows:
      # row is a tuple (key, source_timestamp)
      key_str = row[0]
      source_timestamp = row[1]
      old_value = value.get()
      if old_value is None:
        old_value = 0
      self.value_state.update((old_value + 1,))
      self.map_state.update((old_value,), (key_str,))
      self.list_state.appendValue((key_str,))
      yield Row(key=key_str, value=old_value + 1, timestamp=source_timestamp)

  def close(self) -> None:
    pass


output_schema = StructType(
  [
    StructField("key", StringType(), True),
    StructField("value", LongType(), True),
    StructField("timestamp", TimestampType(), True),
  ]
)

query = (
  spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", broker_address)
  .option("subscribe", input_topic)
  .load()
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp")
  .groupBy("key")
  .transformWithState(
    statefulProcessor=RTMStatefulProcessor(),
    outputStructType=output_schema,
    outputMode="Update",
    timeMode="processingTime",
  )
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", broker_address)
  .option("topic", output_topic)
  .option("checkpointLocation", checkpoint_location)
  .trigger(realTime="5 minutes")
  .outputMode("Update")
  .start()
)
nota

Há uma diferença entre o modo de tempo real e outros modos de execução na transmissão estruturada que executam o StatefulProcessor em transformWithState. Consulte Utilizar transformWithState no modo em tempo real.

Pias​

Escrevendo para o Postgres via ForEachSink​

Scala
import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.sql.{ForeachWriter, Row}

/**
 * Groups connection properties for
 * the JDBC writers.
 *
 * @param url JDBC url of the form jdbc:subprotocol:subname to connect to
 * @param dbtable database table that should be written into
 * @param username username for authentication
 * @param password password for authentication
 */
class JdbcWriterConfig(
    val url: String,
    val dbtable: String,
    val username: String,
    val password: String,
) extends Serializable

/**
 * Handles streaming data writes to a database sink via JDBC, by:
 *  - connecting to the database
 *  - buffering incoming data rows in batches to reduce write overhead
 *
 * @param config connection parameters and configuration knobs for the writer
 */
class JdbcStreamingDataWriter(config: JdbcWriterConfig)
  extends ForeachWriter[Row] with Serializable {
  // The writer currently only supports this hard-coded schema
  private val UPSERT_STATEMENT_SQL =
    s"""MERGE INTO "${config.dbtable}"
       |USING (
       |  SELECT
       |    CAST(? AS INTEGER) AS "id",
       |    CAST(? AS CHARACTER VARYING) AS "data"
       |) AS "source"
       |ON "test"."id" = "source"."id"
       |WHEN MATCHED THEN
       |  UPDATE SET "data" = "source"."data"
       |WHEN NOT MATCHED THEN
       |  INSERT ("id", "data") VALUES ("source"."id", "source"."data")
       |""".stripMargin

  private val MAX_BUFFER_SIZE = 3
  private val buffer = new Array[Row](MAX_BUFFER_SIZE)
  private var bufferSize = 0

  private var connection: Connection = _

  /**
   * Flushes the [[buffer]] by writing all rows in the buffer to the database.
   */
  private def flushBuffer(): Unit = {
    require(connection != null)

    if (bufferSize == 0) {
      return
    }

    var upsertStatement: PreparedStatement = null

    try {
      upsertStatement = connection.prepareStatement(UPSERT_STATEMENT_SQL)

      for (i <- 0 until bufferSize) {
        val row = buffer(i)
        upsertStatement.setInt(1, row.getAs[String]("key"))
        upsertStatement.setString(2, row.getAs[String]("value"))
        upsertStatement.addBatch()
      }

      upsertStatement.executeBatch()
      connection.commit()

      bufferSize = 0
    } catch { case e: Exception =>
      if (connection != null) {
        connection.rollback()
      }
      throw e
    } finally {
      if (upsertStatement != null) {
        upsertStatement.close()
      }
    }
  }

  override def open(partitionId: Long, epochId: Long): Boolean = {
    connection = DriverManager.getConnection(config.url, config.username, config.password)
    true
  }

  override def process(row: Row): Unit = {
    buffer(bufferSize) = row
    bufferSize += 1
    if (bufferSize >= MAX_BUFFER_SIZE) {
      flushBuffer()
    }
  }

  override def close(errorOrNull: Throwable): Unit = {
    flushBuffer()
    if (connection != null) {
      connection.close()
      connection = null
    }
  }
}


spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", testUtils.brokerAddress)
      .option("subscribe", inputTopic)
      .load()
      .writeStream
      .outputMode(OutputMode.Update())
      .trigger(defaultTrigger)
      .foreach(new JdbcStreamingDataWriter(new JdbcWriterConfig(jdbcUrl, tableName, jdbcUsername, jdbcPassword)))
      .start()

Exibição​

important

Este recurso está disponível no Databricks Runtime, versão 17.1 e superior.

Fonte da taxa de exibição​

Neste exemplo, você lê de uma fonte de taxa e exibe a transmissão DataFrame em um Notebook.

Python
inputDF = (
  spark
  .readStream
  .format("rate")
  .option("numPartitions", 2)
  .option("rowsPerSecond", 1)
  .load()
)
display(inputDF, realTime="5 minutes", outputMode="update")
Última atualização em