Vývoj schématu v úložišti stavů

Tento článek obsahuje přehled vývoje schématu v úložišti stavů a příklady typů podporovaných změn schématu.

Co je vývoj schématu v úložišti stavů?

Vývoj schématu označuje schopnost aplikace zpracovávat změny schématu dat.

Azure Databricks podporuje vývoj schématu v úložišti stavů RocksDB pro aplikace strukturovaného streamování, které používají transformWithState.

Vývoj schématu poskytuje flexibilitu pro vývoj a snadnou údržbu. Vývoj schématu slouží k přizpůsobení datového modelu nebo datových typů ve vašem úložišti stavů bez ztráty informací o stavu nebo vyžadování úplného opětovného zpracování historických dat.

Požadavky

Pokud chcete použít vývoj schématu, musíte nastavit formát kódování úložiště stavů na Avro. Pokud chcete tuto možnost nastavit pro aktuální relaci, spusťte následující příkaz:

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

Vývoj schématu je podporován pouze pro stavové operace, které používají transformWithState nebo transformWithStateInPandas. Tyto operátory a související rozhraní API a třídy mají následující požadavky:

K dispozici v Databricks Runtime 16.2 a novějších.
Výpočetní prostředky musí používat vyhrazený režim přístupu nebo bez izolace.
Musíte použít poskytovatele úložiště stavů RocksDB. Databricks doporučuje povolit RocksDB jako součást konfigurace výpočetních prostředků.
transformWithStateInPandas podporuje standardní režim přístupu v Databricks Runtime 16.3 a novějších.

Pokud chcete pro aktuální relaci povolit poskytovatele úložiště stavů RocksDB, spusťte následující příkaz:

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

Podporované vzory vývoje schématu v úložišti stavů

Databricks podporuje následující vzory vývoje schématu pro stavové operace strukturovaného streamování.

Vzor	Popis
Rozšíření typu	Změňte datové typy z více omezujících na méně omezující typy.
Přidání polí	Přidejte nová pole do schématu existujících proměnných úložiště stavů.
Odebrání polí	Odeberte existující pole ze schématu nebo proměnné úložiště stavu.
Změna pořadí polí	Změnit pořadí polí v proměnné
Přidání proměnných stavu	Přidejte do aplikace novou stavovou proměnnou.
Odebrání stavových proměnných	Odeberte existující stavovou proměnnou z aplikace.

Kdy dojde k vývoji schématu?

Vývoj schématu v úložišti stavů vede k aktualizaci kódu, který definuje stavovou aplikaci. Z tohoto důvodu platí následující příkazy:

Vývoj schématu se neprojevuje automaticky v důsledku změn schématu ve zdrojových datech dotazu.
Vývoj schématu nastane pouze v případě, že je nasazena nová verze aplikace. Vzhledem k tomu, že současně může běžet pouze jedna verze streamovacího dotazu, je nutné restartovat úlohu streamování, aby se změnilo schéma proměnných stavu.
Váš kód explicitně definuje všechny proměnné stavu a nastaví schéma pro všechny proměnné stavu.
- V jazyce Scala použijete Encoder k určení schématu pro každou proměnnou.
- V Pythonu explicitně vytvoříte schéma jako StructType.

Nepodporované vzorce vývoje schématu

Následující vzory vývoje schématu nejsou podporovány:

Přejmenování pole: Přejmenování polí není podporováno, protože pole se shodují s názvem. Pokus o přejmenování pole se zpracovává odebráním pole a přidáním nového pole. Tato operace nemá za následek chybu, protože je povoleno odebrání a přidání polí, ale hodnoty z původního pole se nepřenesou do nového pole.
Může se změnit přejmenování klíče nebo změna typu: Název nebo typ klíčů v proměnných stavu mapování nelze změnit.
Zúžení typu Operace zužování typu, označované také jako downcasting, nejsou podporovány. Tyto operace můžou vést ke ztrátě dat. Následující jsou příklady nepodporovaných operací upřesnění typu:
- double nelze zúžit na float, longnebo int
- float nelze zúžit na longint
- long nelze zúžit na int

Rozšíření typu v úložišti stavů

Primitivní datové typy můžete rozšířit na vhodnější typy. Následující změny rozšíření typu jsou podporovány:

int mohou být povýšeny na long, floatnebo double
longmohou být povýšeny na floatdouble
float lze zvýšit úroveň na double
string lze zvýšit úroveň na bytes
bytes lze zvýšit úroveň na string

Existující hodnoty se přetypují jako nový typ. Například 12 se stane 12.00.

Příklad rozšíření typu pomocí `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
    }
  }
}

Python

class IntStateProcessor(StatefulProcessor):
    def init(self, handle):
        # Initial schema with Integer field
        state_schema = StructType([
            StructField("value1", IntegerType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            # Convert input value to integer and update state
            value = pdf["value"].iloc[0]
            self.state.update((int(value),))

        # Read current state
        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "stateValue": [current_state[0]]
        })

class LongStateProcessor(StatefulProcessor):
    def init(self, handle):
        # Later schema with Long field (type widening)
        state_schema = StructType([
            StructField("value1", LongType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            # Convert input value to long and update state
            value = pdf["value"].iloc[0]
            # When reading state written with IntStateProcessor,
            # it will be automatically converted to Long
            self.state.update((int(value),))

        # Read current state
        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "stateValue": [current_state[0]]
        })

Přidejte pole do hodnot úložiště stavů

Do schématu existujících hodnot úložiště stavů můžete přidat nová pole.

Při čtení dat zapsaných pomocí starého schématu kodér Avro vrátí data pro přidaná pole nativně kódovaná jako null.

Python tyto hodnoty vždy interpretuje jako None. Scala má jiné výchozí chování v závislosti na typu pole. Databricks doporučuje implementovat logiku, aby se zajistilo, že Scala nenačítá hodnoty chybějících dat. Viz Výchozí hodnoty polí přidaných do stavové proměnné.

Příklady přidání nových polí pomocí `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
    }
  }
}

Python

class StateV1Processor(StatefulProcessor):
    def init(self, handle):
        # Initial schema with a single field
        state_schema = StructType([
            StructField("value1", IntegerType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state.update((int(value),))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "stateValue": [current_state[0]]
        })

class StateV2Processor(StatefulProcessor):
    def init(self, handle):
        # Later schema with additional fields
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]

            # Read current state
            current_state = self.state.get()
            # When reading state written with StateV1(1),
            # it will be automatically converted to StateV2(1, None)
            value1 = current_state[0]
            value2 = current_state[1]

            # Now update with both fields populated
            self.state.update((int(value), f"metadata-{value}"))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]]
        })

Odebrání polí pro ukládání hodnot stavu

Pole můžete ze schématu existující proměnné odebrat. Při čtení dat se starým schématem se pole ve starých datech, ale ne v novém schématu, ignorují.

Příklady odebrání polí z proměnných stavu

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
    }
  }
}

Python

class RemoveFieldsOriginalProcessor(StatefulProcessor):
    def init(self, handle):
        # Initial schema with multiple fields
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state.update((int(value), f"metadata-{value}"))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]]
        })

class RemoveFieldsReducedProcessor(StatefulProcessor):
    def init(self, handle):
        # Later schema with field removed
        state_schema = StructType([
            StructField("value1", IntegerType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]

            # When reading state written with RemoveFieldsOriginalProcessor(1, "metadata-1"),
            # it will be automatically converted to just (1,)
            current_state = self.state.get()
            value1 = current_state[0]

            self.state.update((int(value),))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]]
        })

Změna pořadí polí v proměnné stavu

Můžete změnit pořadí polí ve stavové proměnné, včetně při přidávání nebo odebírání existujících polí. Pole ve stavových proměnných se shodují podle názvu, nikoli pozice.

Příklady změny pořadí polí v proměnné stavu

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
    }
  }
}

Python

class OrderedFieldsProcessor(StatefulProcessor):
    def init(self, handle):
        # Initial schema with fields in original order
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state.update((int(value), f"metadata-{value}"))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]]
        })

class ReorderedFieldsProcessor(StatefulProcessor):
    def init(self, handle):
        # Later schema with reordered fields
        state_schema = StructType([
            StructField("value2", StringType(), True),
            StructField("value1", IntegerType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]

            # When reading state written with OrderedFieldsProcessor(1, "metadata-1"),
            # it will be automatically converted to ("metadata-1", 1)
            current_state = self.state.get()
            value2 = current_state[0]
            value1 = current_state[1]

            self.state.update((f"new-metadata-{value}", int(value)))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value2": [current_state[0]],
            "value1": [current_state[1]]
        })

Přidání stavové proměnné do stavové aplikace

Mezi spuštění dotazu můžeme také přidat proměnné stavu.

Poznámka: Tento model nevyžaduje kodér Avro a je podporován všemi transformWithState aplikacemi.

Příklad přidání stavové proměnné do stavové aplikace

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
    }
  }
}

Python

class MultiStateV1Processor(StatefulProcessor):
    def init(self, handle):
        # Initial schema with a single state variable
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        self.state1 = handle.getValueState("testState1", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state1.update((int(value), f"metadata-{value}"))

        current_state = self.state1.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]]
        })

class MultiStateV2Processor(StatefulProcessor):
    def init(self, handle):
        # Add a second state variable
        state1_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        state2_schema = StructType([
            StructField("value1", StringType(), True),
            StructField("value2", IntegerType(), True)
        ])
        self.state1 = handle.getValueState("testState1", state1_schema)
        self.state2 = handle.getValueState("testState2", state2_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state1.update((int(value), f"metadata-{value}"))

            # Access and update the new state variable
            current_state2 = self.state2.get()  # Will be None on first run
            self.state2.update((f"new-metadata-{value}", int(value)))

        current_state1 = self.state1.get()
        current_state2 = self.state2.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "state1_value1": [current_state1[0]],
            "state1_value2": [current_state1[1]],
            "state2_value1": [current_state2[0]],
            "state2_value2": [current_state2[1]]
        })

Odebrání stavové proměnné ze stavové aplikace

Kromě odebrání polí můžete také odebrat stavové proměnné mezi spuštěními dotazu.

Poznámka: Tento model nevyžaduje kodér Avro a podporuje ho všechny transformWithState aplikace.

Příklad odebrání stavové proměnné do stavové aplikace

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
    }
  }
}

Python

class MultiStateV2Processor(StatefulProcessor):
    def init(self, handle):
        # Add a second state variable
        state1_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        state2_schema = StructType([
            StructField("value1", StringType(), True),
            StructField("value2", IntegerType(), True)
        ])
        self.state1 = handle.getValueState("testState1", state1_schema)
        self.state2 = handle.getValueState("testState2", state2_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state1.update((int(value), f"metadata-{value}"))

            # Access and update the new state variable
            current_state2 = self.state2.get()  # Will be None on first run
            self.state2.update((f"new-metadata-{value}", int(value)))

        current_state1 = self.state1.get()
        current_state2 = self.state2.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "state1_value1": [current_state1[0]],
            "state1_value2": [current_state1[1]],
            "state2_value1": [current_state2[0]],
            "state2_value2": [current_state2[1]]
        })

class RemoveStateVarProcessor(StatefulProcessor):
    def init(self, handle):
        # Only use one state variable and delete the other
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        self.state1 = handle.getValueState("testState1", state_schema)

        # Delete old state variable that we no longer need
        handle.deleteIfExists("testState2")

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state1.update((int(value), f"metadata-{value}"))

        current_state = self.state1.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]]
        })

Výchozí hodnoty pro pole přidaná do stavové proměnné

Při přidávání nových polí do existující stavové proměnné mají proměnné stavu zapsané pomocí starého schématu následující chování:

Kodér Avro vrátí null hodnotu pro přidaná pole.
Python tyto hodnoty převede na None všechny datové typy.
Výchozí chování Scala se liší podle datového typu:
- Odkazové typy vrací null.
- Primitivní typy vracejí výchozí hodnotu, která se liší na základě primitivního typu. Příklady zahrnují 0 pro int typy nebo false pro bool typy.

Neexistuje žádná integrovaná funkce ani metadata, která označují pole jako přidané prostřednictvím vývoje schématu. Logiku musíte implementovat pro zpracování hodnot null vrácených pro pole, která v předchozím schématu neexistovala.

V případě scaly se můžete vyhnout imputování výchozích hodnot pomocí funkce Option[<Type>], která místo použití výchozího typu vrátí chybějící hodnoty None .

Logiku musíte implementovat, abyste správně zvládli situace, kdy jsou vráceny hodnoty typu None kvůli změnám schématu.

Příklad výchozích hodnot pro přidaná pole do stavové proměnné

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
    }
  }
}

Python

class NullDefaultsProcessor(StatefulProcessor):
    def init(self, handle):
        # Initial schema
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]
            self.state.update((int(value), f"metadata-{value}"))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]]
        })

class ExpandedNullDefaultsProcessor(StatefulProcessor):
    def init(self, handle):
        # Evolution: Adding new fields with null/default values
        state_schema = StructType([
            StructField("value1", IntegerType(), True),
            StructField("value2", StringType(), True),
            StructField("value3", LongType(), True),
            StructField("value4", IntegerType(), True),
            StructField("value5", BooleanType(), True)
        ])
        self.state = handle.getValueState("testState", state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            value = pdf["value"].iloc[0]

            # Reading from state
            current_state = self.state.get()

            # Showing how null defaults work in Python
            # When reading state written with NullDefaultsProcessor state = (1, "metadata-1"),
            # it will be automatically converted to (1, "metadata-1", None, None, None)
            # In Python, both primitive and reference types will be None

            value1 = current_state[0]
            value2 = current_state[1]
            value3 = current_state[2]  # Will be None when evolved from older schema
            value4 = current_state[3]  # Will be None when evolved from older schema
            value5 = current_state[4]  # Will be None when evolved from older schema

            # Check if value3 is None
            if value3 is None:
                print("The value3 field is None (default value for evolution)")
                value3 = 100  # Set a real value now

            # Now update with all fields populated
            self.state.update((
                value1,
                value2,
                value3,
                value4 if value4 is not None else 42,
                value5 if value5 is not None else True
            ))

        current_state = self.state.get()

        yield pd.DataFrame({
            "id": [key[0]],
            "value1": [current_state[0]],
            "value2": [current_state[1]],
            "value3": [current_state[2]],
            "value4": [current_state[3]],
            "value5": [current_state[4]]
        })

Omezení

Následující tabulka popisuje výchozí omezení změn vývoje schématu:

Popis	Výchozí omezení	Konfigurace Sparku k přepsání nastavení
Vývoj schémat pro proměnnou stavu Použití více změn schématu v restartování dotazu se počítá jako vývoj jednoho schématu.	16	`spark.sql.streaming.stateStore.valueStateSchemaEvolutionThreshold`
Vývoj schémat pro dotaz streamování Použití více změn schématu v restartování dotazu se počítá jako vývoj jednoho schématu.	128	`spark.sql.streaming.stateStore.maxNumStateSchemaFiles`

Při řešení potíží s vývojem schématu pro stavové proměnné pečlivě zvažte následující podrobnosti:

Některé vzory nejsou podporovány pro vývoj schématu. Viz Nepodporované vzory vývoje schématu.
Vývoj schématu splňuje všechny požadavky transformWithState a vyžaduje kódovací formát Avro. Viz Požadavky.
Pokud chcete nasadit změny kódu, které vedou k vývoji schématu, musíte restartovat streamovací dotaz. Podívejte se , kdy dochází k vývoji schématu?.

Váš názor

Byla tato stránka užitečná?

Last updated on 2025-04-03

Sdílet prostřednictvím

Vývoj schématu v úložišti stavů

Co je vývoj schématu v úložišti stavů?

Požadavky

Podporované vzory vývoje schématu v úložišti stavů

Kdy dojde k vývoji schématu?

Nepodporované vzorce vývoje schématu

Rozšíření typu v úložišti stavů

Příklad rozšíření typu pomocí transformWithState

Scala

Python

Přidejte pole do hodnot úložiště stavů

Příklady přidání nových polí pomocí transformWithState

Scala

Python

Odebrání polí pro ukládání hodnot stavu

Příklady odebrání polí z proměnných stavu

Scala

Python

Změna pořadí polí v proměnné stavu

Příklady změny pořadí polí v proměnné stavu

Scala

Python

Přidání stavové proměnné do stavové aplikace

Příklad přidání stavové proměnné do stavové aplikace

Scala

Python

Odebrání stavové proměnné ze stavové aplikace

Příklad odebrání stavové proměnné do stavové aplikace

Scala

Python

Výchozí hodnoty pro pole přidaná do stavové proměnné

Příklad výchozích hodnot pro přidaná pole do stavové proměnné

Scala

Python

Omezení

Váš názor

Další materiály

Příklad rozšíření typu pomocí `transformWithState`

Příklady přidání nových polí pomocí `transformWithState`