Pular para o conteúdo principal

evolução do esquema no armazenamento do estado

Este artigo fornece uma visão geral da evolução do esquema no armazenamento do estado e exemplos de tipos de alterações de esquema suportadas.

O que é a evolução do esquema no armazenamento do estado?​

A evolução do esquema refere-se à capacidade de um aplicativo de lidar com alterações no esquema de dados.

Databricks suporta a evolução do esquema no RocksDB armazenamento do estado para aplicativos de transmissão estruturada que usam transformWithState.

A evolução do esquema oferece flexibilidade para o desenvolvimento e facilidade de manutenção. Use a evolução do esquema para adaptar o modelo de dados ou os tipos de dados em seu armazenamento do estado sem perder as informações do estado ou exigir o reprocessamento completo dos dados históricos.

Requisitos​

O senhor deve definir o formato de codificação do armazenamento do estado como Avro para usar a evolução do esquema. Para definir isso para a sessão atual, execute o seguinte:

Python
spark.conf.set("spark.sql.streaming.stateStore.encodingFormat", "avro")

A evolução do esquema é suportada somente para operações com estado que usam transformWithState ou transformWithStateInPandas. Esses operadores e as APIs e classes relacionadas têm os seguintes requisitos:

  • Disponível em Databricks Runtime 16.2 e acima.
  • deve usar o modo de acesso dedicado ou sem isolamento.
  • O senhor deve usar o provedor RocksDB armazenamento do estado. Databricks recomenda ativar o RocksDB como parte da configuração do compute.
  • transformWithStateInPandas suporta o modo de acesso padrão em Databricks Runtime 16.3 e acima.

Para habilitar o provedor RocksDB armazenamento do estado para a sessão atual, execute o seguinte:

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

Apoio à evolução dos padrões de esquema no armazenamento do estado​

Databricks suporta os seguintes padrões de evolução do esquema para operações de transmissão estruturada com estado.

Padrão

Descrição

Ampliação de tipo

Altere os tipos de dados de tipos mais restritivos para menos restritivos.

Adicionando campos

Adicionar novos campos ao esquema de variáveis de armazenamento do estado existentes.

Removendo campos

Remover campos existentes do esquema ou de uma variável de armazenamento do estado.

Reordenando campos

Reordene os campos em uma variável.

Adicionando variáveis de estado

Adicione uma nova variável de estado a um aplicativo.

Removendo variáveis de estado

Remova uma variável de estado existente de um aplicativo.

Quando ocorre a evolução do esquema?​

A evolução do esquema no armazenamento do estado resulta da atualização do código que define seu aplicativo com estado. Por isso, as seguintes afirmações se aplicam:

  • A evolução do esquema não ocorre automaticamente como resultado de alterações de esquema nos dados de origem da consulta.
  • A evolução do esquema ocorre somente quando uma nova versão do aplicativo é implantada. Como apenas uma versão de uma consulta de transmissão pode ser executada simultaneamente, o senhor deve reiniciar o Job de transmissão para evoluir o esquema das variáveis de estado.
  • Seu código define explicitamente todas as variáveis de estado e define o esquema para todas as variáveis de estado.
    • Em Scala, o senhor usa um Encoder para especificar o esquema de cada variável.
    • Em Python, o senhor constrói explicitamente um esquema como StructType.

Evolução não suportada dos padrões de esquema​

Os seguintes padrões de evolução do esquema não são suportados:

  • Renomeação de campos : a renomeação de campos não é suportada porque os campos são correspondidos por nome. A tentativa de renomear um campo é feita removendo o campo e adicionando um novo campo. Essa operação não resulta em um erro, pois é permitido remover e adicionar campos, mas os valores do campo original não são transferidos para o novo campo.

  • Pode key renomear ou alterar o tipo : O senhor não pode alterar o nome ou o tipo de chave nas variáveis de estado do mapa.

  • Restrição de tipo As operações de restrição de tipo, também conhecidas como downcasting , não são compatíveis. Essas operações podem resultar em perda de dados. A seguir, exemplos de operações de restrição de tipo sem suporte:

    • double não pode ser reduzido para float, long ou int
    • float não pode ser reduzido para long ou int
    • long não pode ser reduzido a int

Ampliação de tipos no armazenamento do estado​

Você pode ampliar os tipos de dados primitivos para tipos mais flexíveis. As seguintes alterações de ampliação de tipo são suportadas:

  • int pode ser promovido para long, float ou double
  • long pode ser promovido para float ou double
  • float pode ser promovido a double
  • string pode ser promovido a bytes
  • bytes pode ser promovido a string

Os valores existentes são atualizados como o novo tipo. Por exemplo, 12 se torna 12.00.

Exemplo de ampliação de tipo com transformWithState​

Scala
// Initial run with Integer field
case class StateV1(value1: Integer)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV1](
      "testState",
      Encoders.product[StateV1],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state.update(StateV1(value.toInt))
      value
    }
  }
}

// Later run with Long field (type widening)
case class StateV2(value1: Long)

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV2](
      "testState",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state.update(StateV2(value.toLong))
      value
    }
  }
}

Adicionar campos aos valores de armazenamento do estado​

O senhor pode adicionar novos campos ao esquema de valores de armazenamento do estado existentes.

Ao ler dados gravados com o esquema antigo, o codificador Avro retorna dados para campos adicionados codificados nativamente como null.

O Python sempre interpreta esses valores como None. Scala tem um comportamento diferente no site default, dependendo do tipo do campo. A Databricks recomenda a implementação de lógica para garantir que o Scala não impute valores para dados ausentes. Veja os valores padrão para os campos adicionados à variável de estado.

Exemplos de adição de novos campos com transformWithState​

Scala
// Initial run with single field
case class StateV1(value1: Integer)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV1](
      "testState",
      Encoders.product[StateV1],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state.update(StateV1(value.toInt))
      value
    }
  }
}

// Later run with additional field
case class StateV2(value1: Integer, value2: String)

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV2](
      "testState",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      // When reading state written with StateV1(1),
      // it will be automatically converted to StateV2(1, null)
      val currentState = state.get()
      // Now update with both fields populated
      state.update(StateV2(value.toInt, s"metadata-${value}"))
      value
    }
  }
}

Remover campos para valores de armazenamento do estado​

Você pode remover campos do esquema de uma variável existente. Ao ler dados com o esquema antigo, os campos presentes nos dados antigos, mas não no novo esquema, são ignorados.

Exemplos de remoção de campos de variáveis de estado​

Scala
// Initial run with multiple fields
case class StateV1(value1: Integer, value2: String)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV1](
      "testState",
      Encoders.product[StateV1],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state.update(StateV1(value.toInt, s"metadata-${value}"))
      value
    }
  }
}

// Later run with field removed
case class StateV2(value1: Integer)

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV2](
      "testState",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      // When reading state written with StateV1(1, "metadata-1"),
      // it will be automatically converted to StateV2(1)
      val currentState = state.get()
      state.update(StateV2(value.toInt))
      value
    }
  }
}

Reordenar campos em uma variável de estado​

Você pode reordenar campos em uma variável de estado, inclusive ao adicionar ou remover campos existentes. Os campos nas variáveis de estado são correspondidos por nome, não por posição.

Exemplos de reordenação de campos em uma variável de estado​

Scala
// Initial run with fields in original order
case class StateV1(value1: Integer, value2: String)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV1](
      "testState",
      Encoders.product[StateV1],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state.update(StateV1(value.toInt, s"metadata-${value}"))
      value
    }
  }
}

// Later run with reordered fields
case class StateV2(value2: String, value1: Integer)

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV2](
      "testState",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      // When reading state written with StateV1(1, "metadata-1"),
      // it will be automatically converted to StateV2("metadata-1", 1)
      val currentState = state.get()
      state.update(StateV2(s"new-metadata-${value}", value.toInt))
      value
    }
  }
}

Adicionar uma variável de estado a um aplicativo com estado​

Também podemos adicionar variáveis de estado entre a execução da consulta.

Observação : esse padrão não requer um codificador Avro e é compatível com todos os aplicativos transformWithState.

Exemplo de adição de uma variável de estado a um aplicativo com estado​

Scala
// Initial run with fields in original order
case class StateV1(value1: Integer, value2: String)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state1: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state1 = getHandle.getValueState[StateV1](
      "testState1",
      Encoders.product[StateV1],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state1.update(StateV1(value.toInt, s"metadata-${value}"))
      value
    }
  }
}

case class StateV2(value1: String, value2: Integer)

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state1: ValueState[StateV1] = _
  @transient var state2: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state1 = getHandle.getValueState[StateV1](
      "testState1",
      Encoders.product[StateV1],
      TTLConfig.NONE)
    state2 = getHandle.getValueState[StateV2](
      "testState2",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state1.update(StateV1(value.toInt, s"metadata-${value}"))
      val currentState2 = state2.get()
      state2.update(StateV2(s"new-metadata-${value}", value.toInt))
      value
    }
  }
}

Remover uma variável de estado de um aplicativo com estado​

Além de remover campos, o senhor também pode remover variáveis de estado entre a execução da consulta.

Observação : esse padrão não requer um codificador Avro e é compatível com todos os aplicativos transformWithState.

Exemplo de remoção de uma variável de estado para um aplicativo com estado​

Scala
case class StateV1(value1: Integer, value2: String)
case class StateV2(value1: Integer, value2: String)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state1: ValueState[StateV1] = _
  @transient var state2: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state1 = getHandle.getValueState[StateV1](
      "testState1",
      Encoders.product[StateV1],
      TTLConfig.NONE)
    state2 = getHandle.getValueState[StateV2](
      "testState2",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state1.update(StateV1(value.toInt, s"metadata-${value}"))
      val currentState2 = state2.get()
      state2.update(StateV2(value.toInt, s"new-metadata-${value}"))
      value
    }
  }
}

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state1: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state1 = getHandle.getValueState[StateV1](
      "testState1",
      Encoders.product[StateV1],
      TTLConfig.NONE)
    // delete old state variable that we no longer need
    getHandle.deleteIfExists("testState2")
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state1.update(StateV1(value.toInt, s"metadata-${value}"))
      value
    }
  }
}

valores padrão para campos adicionados à variável de estado​

Quando você adiciona novos campos a uma variável de estado existente, as variáveis de estado escritas usando o esquema antigo têm o seguinte comportamento:

  • O codificador Avro retorna um valor null para os campos adicionados.
  • O Python converte esses valores em None para todos os tipos de dados.
  • Scala default O comportamento difere de acordo com o tipo de dados:
    • Os tipos de referência retornam null.
    • Os tipos primitivos retornam um valor default, que difere de acordo com o tipo primitivo. Os exemplos incluem 0 para tipos int ou false para tipos bool.

Não há funcionalidade integrada ou metadados que sinalizem que o campo foi adicionado por meio da evolução do esquema. Você deve implementar a lógica para lidar com valores nulos retornados para campos que não existiam em seu esquema anterior.

Para Scala, o senhor pode evitar a imputação dos valores de default usando Option[<Type>], que retorna valores ausentes como None, em vez de usar o tipo default.

O senhor deve implementar a lógica para tratar corretamente as situações em que os valores do tipo None são retornados devido à evolução do esquema.

Exemplo de default valores para campos adicionados a uma variável de estado​

Scala
// Example demonstrating how null defaults work in schema evolution

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

// Initial schema that will be evolved
case class StateV1(value1: Integer, value2: String)

class ProcessorV1 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV1] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV1](
      "testState",
      Encoders.product[StateV1],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      state.update(StateV1(value.toInt, s"metadata-${value}"))
      value
    }
  }
}

// Evolution: Adding a new field with null/default values
case class StateV2(value1: Integer, value2: String, value3: Long, value4: Option[Long])

class ProcessorV2 extends StatefulProcessor[String, String, String] {
  @transient var state: ValueState[StateV2] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    state = getHandle.getValueState[StateV2](
      "testState",
      Encoders.product[StateV2],
      TTLConfig.NONE)
  }

  override def handleInputRows(
    key: String,
    inputRows: Iterator[String],
    timerValues: TimerValues): Iterator[String] = {
    rows.map { value =>
      // Reading from state
      val currentState = state.get()

      // Showing how null defaults work for different types
      // When reading state written with StateV1(1, "metadata-1"),
      // it will be automatically converted to StateV2(1, "metadata-1", 0L, None)
      println(s"Current state: $currentState")

      // For primitive types like Long, the UnsafeRow default for null is 0
      val longValue = if (currentState.value3 == 0L) {
        println("The value3 field is the default value (0)")
        100L // Set a real value now
      } else {
        currentState.value3
      }

      // Now update with all fields populated
      state.update(StateV2(value.toInt, s"metadata-${value}", longValue))
      value
    }
  }
}

Limitações​

A tabela a seguir descreve os limites do site default para alterações na evolução do esquema:

Descrição

limite padrão

Configuração do Spark para substituir

evolução do esquema para uma variável de estado. A aplicação de várias alterações de esquema em uma reinicialização de consulta conta como uma única evolução do esquema.

16

spark.sql.streaming.stateStore.valueStateSchemaEvolutionThreshold

evolução do esquema para a consulta de transmissão. A aplicação de várias alterações de esquema em uma reinicialização de consulta conta como uma única evolução do esquema.

128

spark.sql.streaming.stateStore.maxNumStateSchemaFiles

Considere cuidadosamente os detalhes a seguir ao solucionar problemas de evolução do esquema para variáveis de estado:

Última atualização em