Crie uma aplicação personalizada com estado com transformWithState

Pode utilizar transformWithState para criar aplicações de streaming com estado e para implementar soluções de baixa latência e em tempo quase real. Com operadores com estado personalizados, pode criar lógica com estado arbitrária que lhe permite criar novos casos de uso operacionais que não são possíveis com o processamento tradicional de Streaming Estruturado.

Requerimentos

Os operadores transformWithState e transformWithStateInPandas têm os seguintes requisitos:

• Disponível no Databricks Runtime 16.2 e superior.
  • Para o modo em tempo real, utilize Databricks Runtime 17.3 LTS ou superior. Ver conceitos do modo em tempo real.
  • Para o modo de acesso padrão, o Python está disponível no Databricks Runtime 16.3 e superiores, e o Scala está disponível no Databricks Runtime 17.3 e superiores.
• O RocksDB é o fornecedor padrão de armazenamento de estado no Databricks Runtime 17.3 e superiores.

  • Para o Databricks Runtime 17.2 e anteriores, tem de configurar o provedor do arquivo de estado RocksDB. O Databricks recomenda ativar o RocksDB na configuração Spark.

    spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")
    

O que é transformWithState?

O operador transformWithState aplica um processador de estado personalizado a uma consulta de streaming estruturado. Você deve implementar um processador com estado personalizado para usar o transformWithState. O Structured Streaming inclui APIs para construir o seu processador com estado usando Python, Scala ou Java.

Use transformWithState para aplicar lógica personalizada a uma chave de agrupamento. O seguinte descreve o design de alto nível:

• Defina uma ou mais variáveis de estado.
• A informação de estado mantém-se para cada chave de agrupamento. Pode aceder a cada variável de estado em código definido pelo utilizador.
• Para cada microlote processado, todas as linhas relativas à chave estão disponíveis sob a forma de um iterador.
• Utilize StatefulProcessorHandle com temporizadores e condições definidas pelo utilizador para controlar como emitir linhas.
• Para gerir a expiração e o tamanho do estado, os valores do estado suportam definições individuais de tempo de vida (TTL).

Como transformWithState suporta a evolução de esquemas no armazenamento de estado, pode iterar e atualizar as suas aplicações de produção sem perder a informação histórica do estado. Depois de atualizar o esquema de estado, não é obrigado a reprocessar linhas, o que simplifica a implementação e manutenção do código. Consulte evolução do esquema no repositório estadual.

Definindo um StatefulProcessor

Define-se um processador com estado estendendo a StatefulProcessor classe e implementando os seus métodos.

O Spark passa um StatefulProcessorHandle ao método init do seu StatefulProcessor. Utilize o manípulo para criar variáveis de estado e interagir com o armazenamento de estado.

transformWithState suporta três tipos de estados: ValueState, ListState, e MapState. Cada tipo armazena o estado de cada chave de agrupamento usando uma estrutura de dados subjacente diferente.

Implemente os seguintes métodos para definir a sua lógica personalizada:

• Implemente handleInputRows para controlar como a sua aplicação processa dados, atualiza o estado e emite linhas para cada micro-lote. Consulte Manipular linhas de entrada.
• Implementar handleExpiredTimer para executar lógica baseada no tempo independentemente de a chave de agrupamento receber novas linhas num micro-batch. Veja Gerir temporizadores expirados.
• Opcionalmente, implemente handleInitialState para pré-preencher o estado antes de a sua aplicação processar quaisquer linhas de entrada. Consulte processar o estado inicial.

A tabela seguinte compara os comportamentos funcionais destes métodos:

Podes combinar ambos handleInputRows e handleExpiredTimer para implementar lógica complexa conforme necessário.

Por exemplo, você pode implementar um aplicativo que usa handleInputRows para atualizar valores de estado para cada microlote e definir um temporizador de 10 segundos no futuro. Se não forem processadas linhas adicionais, pode usar handleExpiredTimer para emitir os valores atuais na memória de estados. Se forem processadas novas linhas para a chave de agrupamento, podes limpar o temporizador existente e definir um novo.

StatefulProcessorHandle

No PySpark, a StatefulProcessorHandle classe permite-lhe aceder a funções que controlam como o seu código utiliza a informação de estado.

Ao inicializar um StatefulProcessor, deve sempre importar e passar o StatefulProcessorHandle para a handle variável. A variável handle liga a variável local da sua classe Python à variável de estado.

Tipos de estado personalizados

Você pode implementar vários objetos de estado em um único operador com estado.

Escolha um tipo de estado com base na lógica completa da sua aplicação. Por exemplo, pode registar sessões com um ValueState agrupado por user_id e session_id. Ou, para avaliar condições em múltiplas sessões, use um MapState agrupado por user_id com session_id como chave de mapa.

Se o seu objeto de estado usar um StructType, deve definir nomes únicos para cada campo na estrutura do esquema. Estes nomes são visíveis quando se lê o armazenamento de estado. Consulte Leia as informações de estado do Structured Streaming.

As secções seguintes descrevem os tipos de estados suportados por transformWithState:

ValueState

ValueState Armazena um valor para cada chave de agrupamento.

Um estado de valor pode incluir tipos complexos, como uma estrutura ou tupla. Para ValueState, deve implementar lógica para substituir o valor inteiro.

O tempo de vida de um estado de valor reinicia-se quando o valor é atualizado. Se processares uma chave de origem para ValueState sem atualizares o ValueState armazenado, o TTL não é reiniciado.

ListState

ListState Armazena uma lista para cada chave de agrupamento.

Um estado de lista é uma coleção de valores, cada um dos quais pode incluir tipos complexos. Cada valor numa lista tem o seu próprio tempo de vida.

Você pode adicionar itens a uma lista anexando itens individuais, anexando uma lista de itens ou substituindo a lista inteira por um put. Para repor o tempo de expiração, deve usar a operação put.

MapState

MapState Armazena um mapa para cada chave de agrupamento. Os mapas são o equivalente Apache Spark a um dicionário Python (dict).

Um estado de mapa é uma coleção de chaves distintas que cada uma corresponde a um valor, cada uma das quais pode incluir tipos complexos. Cada par chave-valor num mapa tem o seu próprio tempo de vida.

Podes atualizar o valor de uma chave específica, ou podes remover uma chave e o seu valor. Você pode retornar um valor individual usando sua chave, listar todas as chaves, listar todos os valores ou retornar um iterador para trabalhar com o conjunto completo de pares chave-valor no mapa.

Por exemplo, se a sua consulta usar GROUP BY user_id e quiser definir um mapa para cada session_id, a sua chave de agrupamento é user_id e a MapState chave é session_id:

Python

class SessionTracker(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    self.sessions = handle.getMapState("sessions", "session_id string", "count long")

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    for row in rows:
      session_key = (row["session_id"],)  # session_id is the MapState key
      count = self.sessions.getValue(session_key)[0] if self.sessions.containsKey(session_key) else 0
      new_count = count + 1
      self.sessions.updateValue(session_key, (new_count,))
    yield from []

  def close(self) -> None:
    pass

df.groupBy("user_id").transformWithState(SessionTracker(), ...) # user_id is the grouping key

linguagem de programação Scala

case class Event(userId: String, sessionId: String)

class SessionTracker extends StatefulProcessor[String, Event, (String, Long)] {
  @transient private var sessions: MapState[String, Long] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessions = getHandle.getMapState[String, Long]("sessions", Encoders.STRING, Encoders.scalaLong, TTLConfig.NONE)
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[Event],
      timerValues: TimerValues): Iterator[(String, Long)] = {
    rows.foreach { event =>
      val count = if (sessions.containsKey(event.sessionId)) sessions.getValue(event.sessionId) else 0L
      sessions.updateValue(event.sessionId, count + 1) // sessionId is the MapState key
    }
    Iterator.empty
  }
}

df.as[Event]
  .groupByKey(_.userId) // userId is the grouping key
  .transformWithState(new SessionTracker(), TimeMode.None(), OutputMode.Update())

Crie uma variável de estado personalizada no StatefulProcessor

Ao inicializar o StatefulProcessor, você cria uma variável local para cada objeto de estado que permite interagir com objetos de estado em sua lógica personalizada. Defina e inicialize variáveis de estado sobrescrevendo o método incorporado init na StatefulProcessor classe.

Pode definir qualquer número de objetos de estado usando os métodos getValueState, getListState e getMapState no seu StatefulProcessor.

Cada objeto de estado deve ter o seguinte:

• Um nome único
• Um esquema
  • Em Python, tens de especificar o esquema.
  • No Scala, podes passar um Encoder para especificar o esquema de estados.

Opcionalmente, pode também fornecer uma duração de tempo para viver (TTL) em milissegundos. Se estiver implementando um estado de mapa, você deve fornecer uma definição de esquema separada para as chaves de mapa e os valores.

Use as suas variáveis de estado em métodos com lógica personalizada

Os objetos de estado têm métodos para obter estado, atualizar informação de estado existente e limpar o estado atual.

Cada chave de agrupamento tem informação de estado dedicada.

• O StatefulProcessor emite linhas com base na sua lógica personalizada e no esquema de saída especificado. Ver Emitir linhas.
• Use o leitor statestore para aceder a valores no armazenamento de estado. Este leitor destina-se a cargas de trabalho em lote e não é destinado a cargas de trabalho de baixa latência. Consulte Leia as informações de estado do Structured Streaming.
• A lógica especificada com handleInputRows só é executada se as linhas relativas à chave estiverem presentes num micro-batch. Consulte Manipular linhas de entrada.
• Utilize handleExpiredTimer para implementar lógica baseada no tempo que não dependa da observação de linhas para ser acionada. Veja Gerir temporizadores expirados.

Você pode comparar valores dentro de uma chave de agrupamento. Use um MapState para implementar a lógica com uma segunda chave que sua lógica personalizada pode usar. Por exemplo, agrupar por user_id e usar ip_address como chave de MapState permite monitorizar sessões simultâneas de utilizadores.

Considerações avançadas para trabalhar com o estado

As atualizações de estado são tolerantes a falhas. Se uma tarefa falhar antes de um micro-batch terminar de ser processado, a nova tentativa utiliza o valor do último micro-batch concluído com êxito.

Para um desempenho otimizado, a Databricks recomenda que processe todos os valores no iterador para uma determinada chave e confirme as atualizações numa única operação de escrita. Quando escreves para uma variável de estado, isso desencadeia uma escrita para o RocksDB.

Os valores do estado não têm predefinições. Se a sua lógica exigir ler informação de estado existente, use o exists método.

Para implementar lógica para o estado nulo, MapState as variáveis permitem-lhe verificar chaves individuais ou listar todas as chaves.

Manipular linhas de entrada

Use o handleInputRows método para definir como a sua aplicação processa linhas e atualiza os valores de estado. Este método executa-se cada vez que a sua consulta de Streaming Estruturado processa linhas para uma chave de agrupamento.

Para a maioria dos aplicativos com estado implementados com transformWithState, a lógica central é definida usando handleInputRows.

Para cada atualização de micro-lote processada, todas as linhas do micro-lote correspondentes a uma determinada chave de agrupamento podem ser acedidas através de um iterador. A lógica definida pelo utilizador pode interagir com todas as linhas do microlote atual e com os valores no statestore.

Lidar com temporizadores expirados

Use o handleExpiredTimer método para implementar lógica personalizada com base no tempo decorrido.

Dentro de uma chave de agrupamento, os temporizadores são identificados exclusivamente por seu carimbo de data/hora.

Quando um temporizador expira, o resultado é determinado pela lógica implementada em seu aplicativo. Os padrões comuns incluem:

• Emissão de informações armazenadas em uma variável de estado.
• Expulsão de informações de estado armazenadas.
• Criação de um novo temporizador.

Os temporizadores expirados são acionados mesmo que nenhuma linha relativa à chave associada seja processada num microlote.

Especifique o modo de tempo

Ao passar o seu StatefulProcessor para transformWithState, deve especificar o modo de tempo usando o timeMode parâmetro.

As seguintes opções são suportadas:

eventTimeColumnName

Ao usar o modo de tempo EventTime, o parâmetro eventTimeColumnName especifica o nome da coluna no esquema de saída que contém a marca temporal do evento. O Apache Spark utiliza esta coluna para propagar a marca temporal para o fluxo de saída, permitindo operações subsequentes baseadas no tempo de forma correta.

Python

eventTimeColumnName é um argumento adicional para transformWithState ou transformWithStateInPandas:

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=MyProcessor(),
      outputStructType=output_schema,
      outputMode="Append",
      timeMode="EventTime",
      eventTimeColumnName="outputTimestamp",
    )
    .writeStream...
)

linguagem de programação Scala

transformWithState aceita eventTimeColumnName em vez de timeMode. Esta abordagem utiliza sempre o modo EventTime:

val q = spark
  .readStream
  .format("delta")
  .load(srcDeltaTableDir)
  .as[(String, String)]
  .groupByKey(x => x._1)
  .transformWithState(
    new MyProcessor(),
    "outputTimestamp",
    OutputMode.Append(),
  )
  .writeStream...

Valores de temporizador incorporados

O Databricks recomenda não invocar o relógio do sistema em seu aplicativo com monitoração de estado personalizado, pois isso pode levar a novas tentativas não confiáveis em caso de falha de tarefa. Use os métodos na classe TimerValues quando você deve acessar o tempo de processamento ou marca d'água:

Tempo de vida (TTL) para tipos de estado

Para evitar erros de falta de memória e remover valores de tipo de estado obsoleto, transformWithState suporta um valor opcional de tempo de vida (TTL) para cada valor de tipo de estado. Após a expiração, o TTL expulsa silenciosamente os valores do tipo de estado. O TTL não corre handleExpiredTimer nem qualquer lógica personalizada. Para executar código quando o estado expirar, use um temporizador em vez disso.

Para todos os tipos de estado, o TTL é reiniciado ao atualizar a informação de estado. O TTL é aplicado para cada valor de tipo de estado, com regras diferentes para cada tipo de estado:

• As variáveis de estado têm como escopo agrupar chaves.
• Para ValueState objetos, apenas um único valor é armazenado por chave de agrupamento. TTL aplica-se a este valor.
• Para objetos ListState, a lista pode conter muitos valores. O TTL aplica-se a cada valor numa lista de forma independente.
  • Embora o TTL esteja associado a valores individuais numa variável ListState, a única forma de atualizar um valor individual é com o método put, que substitui todo o conteúdo da variável ListState e repõe o TTL para todos os valores da lista.
• Para objetos MapState, cada chave de mapa tem um valor de estado associado. O TTL aplica-se independentemente a cada par chave-valor num mapa.

Exemplo de aplicação com monitoração de estado

O exemplo seguinte define um processador com estado personalizado, SimpleCounterProcessor, incluindo variáveis de estado de exemplo. SimpleCounterProcessor usa ValueState, ListState, e MapState para contar as linhas de cada chave de agrupamento.

Python (Pandas)

import pandas as pd
from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

spark.conf.set("spark.sql.streaming.stateStore.providerClass","org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

output_schema = StructType(
    [
        StructField("id", StringType(), True),
        StructField("countAsString", StringType(), True),
    ]
)

class SimpleCounterProcessor(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    value_state_schema = StructType([StructField("count", IntegerType(), True)])
    list_state_schema = StructType([StructField("count", IntegerType(), True)])
    self.value_state = handle.getValueState(stateName="valueState", schema=value_state_schema)
    self.list_state = handle.getListState(stateName="listState", schema=list_state_schema)
    # Schema can also be defined using strings and SQL DDL syntax
    self.map_state = handle.getMapState(stateName="mapState", userKeySchema="name string", valueSchema="count int")

  def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
    # Seed the running total from state so the count accumulates across micro-batches
    count = self.value_state.get()[0] if self.value_state.exists() else 0
    for pdf in rows:
      list_state_rows = [(120,), (20,)] # A list of tuples
      self.list_state.put(list_state_rows)
      self.list_state.appendValue((111,))
      self.list_state.appendList(list_state_rows)
      pdf_count = pdf.count()
      count += pdf_count.get("value")
    self.value_state.update((count,)) # Count is passed as a tuple
    iter = self.list_state.get()
    list_state_value = next(iter)[0]
    value = count
    user_key = ("user_key",)
    if self.map_state.exists():
      if self.map_state.containsKey(user_key):
        value += self.map_state.getValue(user_key)[0]
    self.map_state.updateValue(user_key, (value,)) # Value is a tuple
    yield pd.DataFrame({"id": key, "countAsString": str(count)})

q = (df.groupBy("key")
  .transformWithStateInPandas(
    statefulProcessor=SimpleCounterProcessor(),
    outputStructType=output_schema,
    outputMode="Update",
    timeMode="None",
  )
  .writeStream...
)

Python (baseado em linhas)

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

output_schema = StructType(
  [
    StructField("id", StringType(), True),
    StructField("countAsString", StringType(), True),
  ]
)

class SimpleCounterProcessor(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    value_state_schema = StructType([StructField("count", IntegerType(), True)])
    list_state_schema = StructType([StructField("count", IntegerType(), True)])
    self.value_state = handle.getValueState(stateName="valueState", schema=value_state_schema)
    self.list_state = handle.getListState(stateName="listState", schema=list_state_schema)
    self.map_state = handle.getMapState(stateName="mapState", userKeySchema="name string", valueSchema="count int")

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    # Seed the running total from state so the count accumulates across micro-batches
    count = self.value_state.get()[0] if self.value_state.exists() else 0
    for row in rows:
      list_state_rows = [(120,), (20,)]  # A list of tuples
      self.list_state.put(list_state_rows)
      self.list_state.appendValue((111,))
      self.list_state.appendList(list_state_rows)
      count += 1
    self.value_state.update((count,))  # Count is passed as a tuple
    iter_list = self.list_state.get()
    list_state_value = next(iter_list)[0]
    value = count
    user_key = ("user_key",)
    if self.map_state.exists():
      if self.map_state.containsKey(user_key):
        value += self.map_state.getValue(user_key)[0]
    self.map_state.updateValue(user_key, (value,))  # Value is a tuple
    yield Row(id=key[0], countAsString=str(count))

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=SimpleCounterProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream...
)

linguagem de programação Scala

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.{Dataset, Encoder, Encoders , DataFrame}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

spark.conf.set("spark.sql.streaming.stateStore.providerClass","org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

class SimpleCounterProcessor extends StatefulProcessor[String, (String, String), (String, String)] {
  @transient private var countState: ValueState[Int] = _
  @transient private var listState: ListState[Int] = _
  @transient private var mapState: MapState[String, Int] = _

  private val longEncoder = Encoders.scalaLong
  private val intEncoder = Encoders.scalaInt
  private val stringEncoder = Encoders.STRING

  override def init(
      outputMode: OutputMode,
      timeMode: TimeMode): Unit = {
    countState = getHandle.getValueState[Int]("countState",
      intEncoder, TTLConfig.NONE)
    listState = getHandle.getListState[Int]("listState",
      intEncoder, TTLConfig.NONE)
    mapState = getHandle.getMapState[String, Int]("mapState",
      stringEncoder, intEncoder, TTLConfig.NONE)
  }

  override def handleInputRows(
      key: String,
      inputRows: Iterator[(String, String)],
      timerValues: TimerValues): Iterator[(String, String)] = {
    var count = countState.getOption().getOrElse(0)
    for (row <- inputRows) {
      val listData = Array(120, 20)
      listState.put(listData)
      listState.appendValue(count)
      listState.appendList(listData)
      count += 1
    }
    val iter = listState.get()
    var listStateValue = 0
    if (iter.hasNext) {
      listStateValue = iter.next()
    }
    countState.update(count)
    var value = count
    val userKey = "userKey"
    if (mapState.exists()) {
      if (mapState.containsKey(userKey)) {
        value += mapState.getValue(userKey)
      }
    }
    mapState.updateValue(userKey, value)
    Iterator((key, count.toString))
  }
}

val q = spark
        .readStream
        .format("delta")
        .load("$srcDeltaTableDir")
        .as[(String, String)]
        .groupByKey(x => x._1)
        .transformWithState(
            new SimpleCounterProcessor(),
            TimeMode.None(),
            OutputMode.Update(),
        )
        .writeStream...

Execute o exemplo de ponta a ponta

O processador acima define a lógica com estado, mas não inicia uma consulta. Para executar SimpleCounterProcessor com copiar e colar, crie inicialmente uma pequena tabela Delta Lake como fonte de streaming e, em seguida, inicie uma consulta que escreve para um destino em memória. Este exemplo utiliza Trigger.AvailableNow para que a consulta processe as linhas pré-carregadas e pare. Para popular a origem e iniciar a consulta, execute o seguinte:

import uuid

# Create a dedicated schema for the example tables
spark.sql("CREATE SCHEMA IF NOT EXISTS main.stateful_examples")

# Seed a small Delta table to use as the streaming source
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.tws_counter_source")
spark.createDataFrame(
  [("a", "1"), ("a", "2"), ("a", "3"), ("b", "1"), ("b", "2")],
  "key string, value string",
).write.saveAsTable("main.stateful_examples.tws_counter_source")

df = spark.readStream.table("main.stateful_examples.tws_counter_source")

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=SimpleCounterProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream.format("memory")
    .queryName("counter_output")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

Depois de concluída a consulta, veja a contagem para cada chave de agrupamento:

display(spark.sql("SELECT id, countAsString FROM counter_output ORDER BY id"))

A a chave tem três linhas e a b chave tem duas, pelo que a consulta devolve:

id  countAsString
a   3
b   2

Para obter mais exemplos, consulte Exemplos de aplicativos com estado.

current_value_tuple = value_state.get() # Returns the value state as a tuple
current_value = current_value_tuple[0]  # Extracts the first item in the tuple
new_value = current_value + 1           # Calculate a new value
value_state.update((new_value,))        # Pass the new value formatted as a tuple

Linhas de emissão

Deve usar handleInputRows ou handleExpiredTimer para definir como transformWithState emite linhas para cada chave de agrupamento. Veja Linhas de entrada do Handle e Temporizadores expirados do Handle.

Aplicações com estado personalizadas não fazem suposições sobre como usar a informação de estado. Para uma determinada condição, a aplicação pode não emitir nenhuma linha, uma linha ou muitas linhas.

Python (Pandas)

Com transformWithStateInPandas, defina o seu esquema de saída com a outputStructType palavra-chave.

Emitir linhas usando um objeto DataFrame pandas e yield.

Opcionalmente, pode yield criar um DataFrame vazio. Se usar update o modo de saída e emitir um DataFrame vazio, isto atualiza os valores da chave de agrupamento para null.

Python (baseado em linhas)

Com transformWithState, defina o seu esquema de saída com a outputStructType palavra-chave.

Gerar linhas utilizando um objeto Row e yield.

Opcionalmente, podes devolver um iterador vazio. Se usar o modo de saída update e emitir um iterador vazio, isto faz com que os valores da chave de agrupamento sejam null.

linguagem de programação Scala

No Scala, emites linhas usando um Iterator objeto. O esquema deriva-se automaticamente a partir do esquema das linhas emitidas.

Opcionalmente, pode devolver um elemento vazio Iterator. Se usar o modo de saída update e emitir um Iterator vazio, isto faz com que os valores da chave de agrupamento sejam null.

Controlar o estado inicial

Opcionalmente, pode passar um estado inicial ao primeiro micro-lote.

Por exemplo, pode usar isto para:

• Migre um fluxo de trabalho existente para uma nova aplicação personalizada.
• Atualizar um operador com estado para alterar o seu esquema ou a sua lógica.
• Repara uma falha que não pode ser reparada automaticamente e que requer intervenção manual.

Você fornece um estado inicial usando um DataFrame com o mesmo esquema de chave de agrupamento que as linhas de entrada.

O exemplo seguinte gera um contador por chave a partir de uma tabela Delta existente:

Python (baseado em linhas)

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

class CounterWithInitialState(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    state_schema = StructType([StructField("count", IntegerType(), True)])
    self.count_state = handle.getValueState("countState", state_schema)

  def handleInitialState(self, key, initialState: Row, timerValues) -> None:
    self.count_state.update((initialState["count"],))

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    count = self.count_state.get()[0] if self.count_state.exists() else 0
    for _ in rows:
      count += 1
    self.count_state.update((count,))
    yield Row(id=key[0], count=count)

  def close(self) -> None:
    pass

output_schema = StructType([
  StructField("id", StringType(), True),
  StructField("count", IntegerType(), True),
])

import uuid

# Create a dedicated schema for the example tables
spark.sql("CREATE SCHEMA IF NOT EXISTS main.stateful_examples")

# Seed existing per-key counts to load as the initial state
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.existing_counts")
spark.createDataFrame(
  [("x", 10)],
  "id string, count int",
).write.saveAsTable("main.stateful_examples.existing_counts")

# Seed a small Delta table to use as the streaming source
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.tws_initial_source")
spark.createDataFrame(
  [("x", "a"), ("x", "b")],
  "id string, value string",
).write.saveAsTable("main.stateful_examples.tws_initial_source")

df = spark.readStream.table("main.stateful_examples.tws_initial_source")

# Load existing counts as initial state — must use the same grouping key as the input
initial_state = spark.read.table("main.stateful_examples.existing_counts").groupBy("id")

q = (
  df.groupBy("id")
    .transformWithState(
      statefulProcessor=CounterWithInitialState(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
      initialState=initial_state,
    )
    .writeStream.format("memory")
    .queryName("initial_state_output")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

# The initial state seeds "x" with 10, and the source adds two rows, so the count is 12
display(spark.sql("SELECT id, count FROM initial_state_output ORDER BY id"))

linguagem de programação Scala

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.Encoders

class CounterWithInitialState
    extends StatefulProcessorWithInitialState[String, (String, String), (String, String), (String, Int)] {

  @transient private var countState: ValueState[Int] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    countState = getHandle.getValueState[Int]("countState", Encoders.scalaInt, TTLConfig.NONE)
  }

  override def handleInitialState(
      key: String, initialState: (String, Int), timerValues: TimerValues): Unit = {
    countState.update(initialState._2)
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[(String, String)],
      timerValues: TimerValues): Iterator[(String, String)] = {
    val count = if (countState.exists()) countState.get() else 0
    val newCount = count + rows.size
    countState.update(newCount)
    Iterator((key, newCount.toString))
  }
}

// Load existing counts as initial state — must use the same grouping key as the input
val initialState = spark.read.table("existing_counts")
  .as[(String, Int)]
  .groupByKey(_._1)

val q = spark
  .readStream
  .format("delta")
  .load(srcDeltaTableDir)
  .as[(String, String)]
  .groupByKey(_._1)
  .transformWithState(
    new CounterWithInitialState(),
    TimeMode.None(),
    OutputMode.Update(),
    initialState,
  )
  .writeStream...

Processamento assíncrono (Beta)

Python transformWithState suporta o processamento assíncrono usando asyncio para executar operações de estado e lógica de utilizador em simultâneo. O processamento assíncrono tem maior rendimento do que o processamento síncrono e requer apenas pequenas alterações de código, sem quaisquer bibliotecas assíncronas de terceiros. Para usar processamento assíncrono, implemente um AsyncStatefulProcessor em vez do síncrono StatefulProcessor. Ver Processamento assíncrono com transformWithState (Beta).

Utilizar transformWithState em pipelines do Lakeflow

Use o transformWithState operador dentro dos pipelines Lakeflow para implementar lógica com estado arbitrário nos seus pipelines de streaming usando Python.

Para fazer isso, conclua as seguintes etapas:

1. Define o esquema de saída e a lógica do processador com estado para as tuas transformações de estado arbitrárias. Para exemplos, veja Exemplos de aplicações com estado.
2. Crie um fluxo do pipeline Lakeflow que invoque o operador transformWithState num DataFrame. Veja o Tutorial: Crie o seu primeiro pipeline usando o Lakeflow Pipelines Editor.
3. Executa o teu pipeline e valida os resultados na tabela de destino ou no sink.

Para um exemplo que usa transformWithState para monitorizar batimentos cardíacos dos sensores, veja Exemplo: Usar transformWithState para monitorizar batimentos cardíacos dos sensores.