Crie um aplicativo personalizado com gerenciamento de estado com transformWithState

Você pode usar transformWithState para criar aplicativos de streaming com estado e implementar soluções de baixa latência e quase em tempo real. Com operadores com estado personalizados, você pode criar uma lógica com estado arbitrário que permite criar novos casos de uso operacional que não são possíveis com o processamento de Streaming Estruturado tradicional.

Requisitos

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, use o Databricks Runtime 17.3 LTS ou superior. Veja conceitos de modo em tempo real.
  • Para o modo de acesso padrão, Python está disponível no Databricks Runtime 16.3 e superior, e o Scala está disponível no Databricks Runtime 17.3 e superior.
• O RocksDB é o provedor de armazenamento de estado padrão no Databricks Runtime 17.3 e superior.

  • Para o Databricks Runtime 17.2 e abaixo, você deve configurar o provedor de repositório de estado do RocksDB. O Databricks recomenda habilitar o RocksDB na configuração do Spark.

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

O que é transformWithState?

O transformWithState operador aplica um processador com estado personalizado a uma consulta de Streaming Estruturado. Você deve implementar um processador com estado personalizado para usar transformWithState. O Structured Streaming inclui APIs para construir seu processador stateful 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.
• As informações de estado persistem para cada chave de agrupamento. Você pode acessar cada variável de estado no código definido pelo usuário.
• Para cada microlote processado, todas as linhas associadas à chave ficam disponíveis na forma de um iterador.
• Use o StatefulProcessorHandle com temporizadores e condições definidas pelo usuário para controlar como as linhas são emitidas.
• Para gerenciar a expiração de estado e o tamanho do estado, os valores de estado dão suporte a definições de TTL (vida útil individual).

Como transformWithState dá suporte à evolução do esquema no repositório de estado, você pode iterar e atualizar seus aplicativos de produção sem perder informações de estado histórico. Depois de atualizar o esquema de estado, você não precisará reprocessar linhas, o que simplifica as implantações de código e a manutenção. Confira a evolução do esquema no armazenamento de estado.

Definindo um StatefulProcessor

Você define um processador com estado estendendo a StatefulProcessor classe e implementando seus métodos.

O Spark passa um StatefulProcessorHandle para o método init da sua StatefulProcessor. Use o handle para criar variáveis de estado e interagir com o armazenamento de estado.

transformWithState dá suporte a três tipos de estado: ValueState, ListStatee MapState. Cada tipo armazena o estado para cada chave de agrupamento usando uma estrutura de dados subjacente diferente.

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

• Implemente handleInputRows para controlar como seu aplicativo processa dados, atualiza o estado e emite linhas para cada microlote. Veja Gerenciar linhas de entrada.
• Implemente handleExpiredTimer para executar a lógica baseada em tempo, independentemente de a chave de agrupamento receber novas linhas em um microlote. Consulte Como lidar com temporizadores expirados.
• Opcionalmente, implemente handleInitialState para preencher previamente o estado antes que seu aplicativo processe as linhas de entrada. Consulte Manipular o estado inicial.

A tabela a seguir compara os comportamentos funcionais desses métodos:

Você pode combinar handleInputRows e handleExpiredTimer para implementar uma 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 nenhuma linha adicional for processada, você poderá usar handleExpiredTimer para emitir os valores atuais no repositório de estado. Se novas linhas forem processadas para a chave de agrupamento, você poderá limpar o temporizador existente e definir um novo temporizador.

StatefulProcessorHandle

No PySpark, a StatefulProcessorHandle classe permite que você acesse funções que controlam como seu código usa informações de estado.

Ao inicializar um StatefulProcessor, você deve sempre importar e passar o StatefulProcessorHandle para a variável handle. A variável handle vincula a variável local em 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 do aplicativo. Por exemplo, você pode acompanhar sessões usando um ValueState agrupado por user_id e session_id. Ou, para avaliar as condições em várias sessões, use um MapState agrupado por user_id com session_id como a chave do mapa.

Se o objeto de estado usar um StructType, você deverá definir nomes exclusivos para cada campo no struct para o esquema. Esses nomes são visíveis ao consultar o armazenamento de estado. Consulte a seção Ler informações de estado do Streaming Estruturado.

As seções a seguir descrevem os tipos de estado compatíveis com transformWithState:

ValueState

ValueState armazena um valor para cada chave de agrupamento.

Um estado de valor pode incluir tipos complexos, como um struct ou tupla. Para ValueState, você deve implementar a lógica para substituir o valor inteiro.

A vida útil de um estado de valor é redefinida quando o valor é atualizado. Se você processar uma chave de origem para ValueState sem atualizar o ValueState armazenado, o tempo de vida útil não é redefinido.

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 em uma lista tem seu próprio tempo de vida útil.

Você pode adicionar itens a uma lista acrescentando itens individuais, acrescentando uma lista de itens ou substituindo toda a lista com um put. Para redefinir o tempo de vida útil, você deve usar uma operação put.

MapState

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

Um estado de mapa é uma coleção de chaves distintas que cada um mapeia para um valor, cada uma delas pode incluir tipos complexos. Cada par chave-valor em um mapa tem seu próprio tempo de vida útil.

Você pode atualizar o valor de uma chave específica ou remover uma chave e 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 consulta usar GROUP BY user_id e você quiser definir um mapa para cada session_idum, sua chave de agrupamento será user_id e a MapState chave será 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

Scala (linguagem de programação)

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

Criar uma variável de estado personalizada no StatefulProcessor

Ao inicializar sua 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 substituindo o método interno init na StatefulProcessor classe.

Você pode definir qualquer número de objetos de estado usando o getValueState, getListStatee getMapState métodos em seu StatefulProcessor.

Cada objeto de estado deve ter o seguinte:

• Um nome exclusivo
• Um esquema
  • Em Python, você deve especificar o esquema.
  • No Scala, você pode passar um Encoder para especificar o esquema de estado.

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

Usar 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ções de estado existentes e limpar o estado atual.

Cada chave de agrupamento tem informações de estado dedicadas.

• O StatefulProcessor emite linhas com base na sua lógica personalizada e no esquema de saída especificado. Consulte Emit rows.
• Use o statestore leitor para acessar valores no repositório de estado. Esse leitor destina-se a cargas de trabalho em lote e não se destina a cargas de trabalho de baixa latência. Consulte a seção Ler informações de estado do Streaming Estruturado.
• A lógica especificada usando handleInputRows somente será executada se as linhas da chave estiverem presentes em um microlote. Veja Gerenciar linhas de entrada.
• Use handleExpiredTimer para implementar lógica baseada em tempo que não depende de observar linhas para ser acionada. Consulte Como lidar com temporizadores expirados.

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

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-lote concluir o processamento, a repetição usará o valor do último microlote bem-sucedido.

Para um desempenho otimizado, o Databricks recomenda que você processe todos os valores no iterador para uma determinada chave e confirme atualizações em uma única gravação. Quando você grava em uma variável de estado, isso dispara uma gravação no RocksDB.

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

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

Manipular linhas de entrada

Use o handleInputRows método para definir como seu aplicativo processa linhas e atualiza valores de estado. Esse método é executado sempre que a consulta de Streaming Estruturado processa linhas para uma chave de agrupamento.

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

Para cada atualização de microlote processada, todas as linhas do microlote de uma dada chave de agrupamento podem ser acessadas por meio de um iterador. A lógica definida pelo usuário pode interagir com todas as linhas do microbatch atual e com os valores no armazenamento de estado.

Manipular temporizadores expirados

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

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

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

• Emitindo informações armazenadas em uma variável de estado.
• Eliminando informações de estado armazenadas.
• Criando um novo temporizador.

Os temporizadores expirados são disparados mesmo que nenhuma linha da chave associada seja processada em um microlote.

Especificar o modo de hora

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

Há suporte para as seguintes opções:

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 data e hora do evento. O Apache Spark usa essa coluna para propagar a marca d'água para o fluxo de saída, permitindo operações corretas baseadas em tempo downstream.

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

Scala (linguagem de programação)

transformWithState aceita eventTimeColumnName no lugar de timeMode. Essa abordagem usa 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 interno

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

TTL (vida útil) para tipos de estado

Para evitar erros fora de memória e remover valores de tipo de estado obsoletos, transformWithState dá suporte a um valor TTL (tempo de vida útil) opcional para cada valor de tipo de estado. Após a expiração, o TTL remove silenciosamente os valores de tipo de estado. O TTL não executa handleExpiredTimer nem nenhuma lógica personalizada. Para executar o código quando o estado expirar, use um temporizador.

Para todos os tipos de estado, o TTL é redefinido ao atualizar as informações de estado. O TTL é imposto para cada valor de tipo de estado, com regras diferentes para cada tipo de estado:

• As variáveis de estado estão limitadas pelas chaves de agrupamento.
• Para ValueState objetos, apenas um único valor é armazenado por chave de agrupamento. TTL aplica-se a esse valor.
• Para ListState objetos, a lista pode conter muitos valores. A TTL aplica-se a cada valor em uma lista de forma independente.
  • Embora o TTL se aplique a valores individuais em um ListState, a única maneira de atualizar um valor individual é com o método put, que substitui todo o conteúdo da variável ListState e redefine o TTL para todos os valores da lista.
• Para MapState objetos, cada chave de mapa tem um valor de estado associado. O TTL aplica-se independentemente a cada par chave-valor em um mapa.

Exemplo de aplicativo com estado

O exemplo a seguir define um processador com estado personalizado, SimpleCounterProcessorincluindo variáveis de estado de exemplo. SimpleCounterProcessor usa ValueState, ListState e MapState para contar 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 linha)

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

Scala (linguagem de programação)

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 por copiar e colar, crie uma pequena tabela Delta Lake como origem de streaming e, em seguida, inicie uma consulta que grave em um coletor na memória. Este exemplo usa Trigger.AvailableNow para que a consulta processe as linhas e paradas seedadas. 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()

Após a consulta ser concluída, 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, então a consulta retorna:

id  countAsString
a   3
b   2

Para obter mais exemplos, consulte Exemplo 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

Emitir linhas

Você deve usar handleInputRows ou handleExpiredTimer para definir como transformWithState emite linhas para cada chave de agrupamento. Consulte Manipular linhas de entrada e Manipular temporizadores expirados.

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

Python (Pandas)

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

Emita linhas usando um objeto DataFrame Pandas e yield.

Opcionalmente, você pode yield um DataFrame vazio. Se você usar update o modo de saída e emitir um DataFrame vazio, isso atualizará os valores para que a chave de agrupamento seja null.

Python (baseado em linha)

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

Emita linhas usando um Row objeto e yield.

Opcionalmente, você pode retornar um iterador vazio. Se você usar update o modo de saída e emitir um iterador vazio, isso atualizará os valores para que a chave de agrupamento seja null.

Scala (linguagem de programação)

No Scala, você emite linhas usando um Iterator objeto. O esquema deriva-se automaticamente do esquema das linhas emitidas.

Opcionalmente, você pode retornar um vazio Iterator. Se você usar o modo de saída update e emitir um Iterator vazio, isso atualizará os valores da chave de agrupamento para null.

Manipular o estado inicial

Opcionalmente, você pode passar um estado inicial para o primeiro microlote.

Por exemplo, você pode usar isso para:

• Migre um fluxo de trabalho existente para um novo aplicativo personalizado.
• Atualize um operador com estado para alterar seu esquema ou lógica.
• Reparar uma falha que não pode ser reparada automaticamente e 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 a seguir inicializa um contador por chave a partir de uma tabela Delta existente:

Python (baseado em linha)

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

Scala (linguagem de programação)

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 oferece suporte ao processamento assíncrono usando asyncio para executar operações de estado e lógica do usuário concorrentemente. O processamento assíncrono tem maior taxa de transferência do que o processamento síncrono e requer apenas pequenas alterações de código, sem nenhuma biblioteca assíncrona de terceiros. Para usar processamento assíncrono, implemente um AsyncStatefulProcessor em vez do síncrono StatefulProcessor. Veja Processamento assíncrono com transformWithState (Beta).

Usar transformWithState em pipelines do Lakeflow

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

Para isso, conclua as seguintes etapas:

1. Defina o esquema de saída e a lógica de processador com estado para suas transformações arbitrárias com estado. Para obter exemplos, consulte Exemplo de aplicativos com estado.
2. Crie um fluxo de pipeline do Lakeflow que invoque o operador transformWithState em um DataFrame. Confira o Tutorial: Criar seu primeiro pipeline usando o Editor do Lakeflow Pipelines.
3. Execute seu pipeline e valide os resultados na tabela ou coletor de destino.

Para obter um exemplo que usa transformWithState para monitorar pulsações do sensor, consulte Exemplo: Use transformWithState para monitorar pulsações do sensor.