• State Processor API
    • Abstraction
    • Reading State
    • Writing New Savepoints

    State Processor API

    Apache Flink’s State Processor API provides powerful functionality to reading, writing, and modifing savepoints and checkpoints using Flink’s batch DataSet api.This is useful for tasks such as analyzing state for interesting patterns, troubleshooting or auditing jobs by checking for discrepancies, and bootstrapping state for new applications.

    • Abstraction
    • Reading State
    • Writing New Savepoints

    Abstraction

    To understand how to best interact with savepoints in a batch context it is important to have a clear mental model of how the data in Flink state relates to a traditional relational database.

    A database can be thought of as one or more namespaces, each containing a collection of tables.Those tables in turn contain columns whose values have some intrinsic relationship between them, such as being scoped under the same key.

    A savepoint represents the state of a Flink job at a particular point in time which is made up of many operators.Those operators contain various kinds of state, both partitioned or keyed state, and non-partitioned or operator state.

    1. MapStateDescriptor<Integer, Double> CURRENCY_RATES = new MapStateDescriptor<>("rates", Types.INT, Types.DOUBLE);
    3. class CurrencyConverter extends BroadcastProcessFunction<Transaction, CurrencyRate, Transaction> {
    5.   public void processElement(
    6.         Transaction value,
    7.         ReadOnlyContext ctx,
    8.         Collector<Transaction> out) throws Exception {
    10.      Double rate = ctx.getBroadcastState(CURRENCY_RATES).get(value.currencyId);
    11.      if (rate != null) {
    12.         value.amount *= rate;
    13.      }
    14.      out.collect(value);
    15.   }
    17.   public void processBroadcastElement(
    18.         CurrencyRate value,
    19.         Context ctx,
    20.         Collector<Transaction> out) throws Exception {
    21.         ctx.getBroadcastState(CURRENCY_RATES).put(value.currencyId, value.rate);
    22.   }
    23. }
    25. class Summarize extends RichFlatMapFunction<Transaction, Summary> {
    26.   transient ValueState<Double> totalState;
    27.   transient ValueState<Integer> countState;
    29.   public void open(Configuration configuration) throws Exception {
    30.      totalState = getRuntimeContext().getState(new ValueStateDescriptor<>("total", Types.DOUBLE));
    31.      countState = getRuntimeContext().getState(new ValueStateDescriptor<>("count", Types.INT));
    32.   }
    34.   public void flatMap(Transaction value, Collector<Summary> out) throws Exception {
    35.      Summary summary = new Summary();
    36.      summary.total = value.amount;
    37.      summary.count = 1;
    39.      Double currentTotal = totalState.value();
    40.      if (currentTotal != null) {
    41.         summary.total += currentTotal;
    42.      }
    44.      Integer currentCount = countState.value();
    45.      if (currentCount != null) {
    46.         summary.count += currentCount;
    47.      }
    48.      countState.update(summary.count);
    50.      out.collect(summary);
    51.   }
    52. }
    54. DataStream<Transaction> transactions = . . .
    55. BroadcastStream<CurrencyRate> rates = . . .
    56. transactions
    57.   .connect(rates)
    58.   .process(new CurrencyConverter())
    59.   .uid("currency_converter")
    60.   .keyBy(transaction -> transaction.accountId)
    61.   .flatMap(new Summarize())
    62.   .uid("summarize")

    This job contains multiple operators along with various kinds of state.When analyzing that state we can first scope data by its operator, named by setting its uid.Within each operator we can look at the registered states.CurrencyConverter has a broadcast state, which is a type of non-partitioned operator state.In general, there is no relationship between any two elements in an operator state and so we can look at each value as being its own row.Contrast this with Summarize, which contains two keyed states.Because both states are scoped under the same key we can safely assume there exists some relationship between the two values.Therefore, keyed state is best understood as a single table per operator containing one key column along with n value columns, one for each registered state.All of this means that the state for this job could be described using the following pseudo-sql commands.

    1. CREATE NAMESPACE currency_converter;
    3. CREATE TABLE currency_converter.rates (
    4.    value Tuple2<Integer, Double>
    5. );
    7. CREATE NAMESPACE summarize;
    9. CREATE TABLE summarize.keyed_state (
    10.    key   INTEGER PRIMARY KEY,
    11.    total DOUBLE,
    12.    count INTEGER
    13. );

    In general, the savepoint ↔ database relationship can be summarized as:

    1. * A savepoint is a database
    2. * An operator is a namespace named by its uid
    3. * Each operator state represents a single table
    4.     * Each element in an operator state represents a single row in that table
    5. * Each operator containing keyed state has a single “keyed_state” table
    6.     * Each keyed_state table has one key column mapping the key value of the operator
    7.     * Each registered state represents a single column in the table
    8.     * Each row in the table maps to a single key

    Reading State

    Reading state begins by specifiying the path to a valid savepoint or checkpoint along with the StateBackend that should be used to restore the data.The compatability guarantees for restoring state are identical to those when restoring a DataStream application.

    1. ExecutionEnvironment bEnv   = ExecutionEnvironment.getExecutionEnvironment();
    2. ExistingSavepoint savepoint = Savepoint.load(bEnv, "hdfs://path/", new RocksDBStateBackend());
    1. val bEnv      = ExecutionEnvironment.getExecutionEnvironment()
    2. val savepoint = Savepoint.load(bEnv, "hdfs://path/", new RocksDBStateBackend())

    When reading operator state, simply specify the operator uid, state name, and type information.

    1. DataSet<Integer> listState  = savepoint.readListState(
    2.     "my-uid",
    3.     "list-state",
    4.     Types.INT);
    6. DataSet<Integer> unionState = savepoint.readUnionState(
    7.     "my-uid",
    8.     "union-state",
    9.     Types.INT);
    11. DataSet<Tuple2<Integer, Integer>> broadcastState = savepoint.readBroadcastState(
    12.     "my-uid",
    13.     "broadcast-state",
    14.     Types.INT,
    15.     Types.INT);
    1. val listState  = savepoint.readListState(
    2.     "my-uid",
    3.     "list-state",
    4.     Types.INT)
    6. val unionState = savepoint.readUnionState(
    7.     "my-uid",
    8.     "union-state",
    9.     Types.INT)
    11. val broadcastState = savepoint.readBroadcastState(
    12.     "my-uid",
    13.     "broadcast-state",
    14.     Types.INT,
    15.     Types.INT)

    A custom TypeSerializer may also be specified if one was used in the StateDescriptor for the state.

    1. DataSet<Integer> listState = savepoint.readListState(
    2.     "uid",
    3.     "list-state", 
    4.     Types.INT,
    5.     new MyCustomIntSerializer());
    1. val listState = savepoint.readListState(
    2.     "uid",
    3.     "list-state", 
    4.     Types.INT,
    5.     new MyCustomIntSerializer())

    When reading keyed state, users specify a KeyedStateReaderFunction to allow reading arbitrary columns and complex state types such as ListState, MapState, and AggregatingState.This means if an operator contains a stateful process function such as:

    1. public class StatefulFunctionWithTime extends KeyedProcessFunction<Integer, Integer, Void> {
    3.    ValueState<Integer> state;
    5.    @Override
    6.    public void open(Configuration parameters) {
    7.       ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("state", Types.INT);
    8.       state = getRuntimeContext().getState(stateDescriptor);
    9.    }
    11.    @Override
    12.    public void processElement(Integer value, Context ctx, Collector<Void> out) throws Exception {
    13.       state.update(value + 1);
    14.    }
    15. }
    1. public class StatefulFunctionWithTime extends KeyedProcessFunction[Integer, Integer, Void] {
    3.    var state: ValueState[Integer];
    5.    override def open(parameters: Configuration) {
    6.       val stateDescriptor = new ValueStateDescriptor("state", Types.INT);
    7.       state = getRuntimeContext().getState(stateDescriptor);
    8.    }
    10.    override def processElement(value: Integer, ctx: Context, out: Collector[Void]) {
    11.       state.update(value + 1);
    12.    }
    13. }

    Then it can read by defining an output type and corresponding KeyedStateReaderFunction.

    1. class KeyedState {
    2.   Integer key;
    3.   Integer value;
    4. }
    6. class ReaderFunction extends KeyedStateReaderFunction<Integer, KeyedState> {
    7.   ValueState<Integer> state;
    9.   @Override
    10.   public void open(Configuration parameters) {
    11.      ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("state", Types.INT);
    12.      state = getRuntimeContext().getState(stateDescriptor);
    13.   }
    15.   @Override
    16.   public void readKey(
    17.     Integer key,
    18.     Context ctx,
    19.     Collector<KeyedState> out) throws Exception {
    21.      KeyedState data = new KeyedState();
    22.      data.key    = key;
    23.      data.value  = state.value();
    24.      out.collect(data);
    25.   }
    26. }
    28. DataSet<KeyedState> keyedState = savepoint.readKeyedState("my-uid", new ReaderFunction());
    1. case class KeyedState(key: Int, value: Int)
    3. class ReaderFunction extends KeyedStateReaderFunction[Integer, KeyedState] {
    4.   var state: ValueState[Integer];
    6.   override def open(parameters: Configuration) {
    7.      val stateDescriptor = new ValueStateDescriptor("state", Types.INT);
    8.      state = getRuntimeContext().getState(stateDescriptor);
    9.   }
    11.   override def processKey(
    12.     key: Int,
    13.     ctx: Context,
    14.     out: Collector[Keyedstate]) throws Exception {
    16.      val data = KeyedState(key, state.value())
    17.      out.collect(data);
    18.   }
    19. }
    21. val keyedState = savepoint.readKeyedState("my-uid", new ReaderFunction());

    Note: When using a KeyedStateReaderFunction all state descriptors must be registered eagerly inside of open. Any attempt to call RuntimeContext#getState, RuntimeContext#getListState, or RuntimeContext#getMapState will result in a RuntimeException.

    Writing New Savepoints

    State writers are based around the abstraction of Savepoint, where one Savepoint may have many operators and the state for any particular operator is created using a BootstrapTransformation.

    A BootstrapTransformation starts with a DataSet containing the values that are to be written into state.The transformation may be optionally keyed depending on whether or not you are writing keyed or operator state.Finally a bootstrap function is applied depending to the transformation; Flink supplies KeyedStateBootstrapFunction for writing keyed state, StateBootstrapFunction for writing non keyed state, and BroadcastStateBootstrapFunction for writing broadcast state.

    1. public  class Account {
    2.     public int id;
    4.     public double amount;    
    6.     public long timestamp;
    7. }
    9. public class AccountBootstrapper extends KeyedStateBootstrapFunction<Integer, Account> {
    10.     ValueState<Double> state;
    12.     @Override
    13.     public void open(Configuration parameters) {
    14.         ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<>("total",Types.DOUBLE);
    15.         state = getRuntimeContext().getState(descriptor);
    16.     }
    18.     @Override
    19.     public void processElement(Account value, Context ctx) throws Exception {
    20.         state.update(value.amount);
    21.     }
    22. }
    24. ExecutionEnvironment bEnv = ExecutionEnvironment.getExecutionEnvironment();
    26. DataSet<Account> accountDataSet = bEnv.fromCollection(accounts);
    28. BootstrapTransformation<Account> transformation = OperatorTransformation
    29.     .bootstrapWith(accountDataSet)
    30.     .keyBy(acc -> acc.id)
    31.     .transform(new AccountBootstrapper());
    1. case class Account(id: Int, amount: Double, timestamp: Long)
    3. class AccountBootstrapper extends KeyedStateBootstrapFunction[Integer, Account] {
    4.     var state: ValueState[Double]
    6.     override def open(parameters: Configuration): Unit = {
    7.         val descriptor = new ValueStateDescriptor[Double]("total",Types.DOUBLE)
    8.         state = getRuntimeContext().getState(descriptor)
    9.     }
    11.     @throws[Exception]
    12.     override def processElement(value: Account, ctx: Context): Unit = {
    13.         state.update(value.amount)
    14.     }
    15. }
    17. val bEnv = ExecutionEnvironment.getExecutionEnvironment()
    19. val accountDataSet = bEnv.fromCollection(accounts)
    21. val transformation = OperatorTransformation
    22.     .bootstrapWith(accountDataSet)
    23.     .keyBy(acc -> acc.id)
    24.     .transform(new AccountBootstrapper())

    The KeyedStateBootstrapFunction supports setting event time and processing time timers.The timers will not fire inside the bootstrap function and only become active once restored within a DataStream application.If a processing time timer is set but the state is not restored until after that time has passed, the timer will fire immediatly upon start.

    Once one or more transformations have been created they may be combined into a single Savepoint. Savepoint’s are created using a state backend and max parallelism, they may contain any number of operators.

    1. Savepoint
    2.     .create(backend, 128)
    3.     .withOperator("uid1", transformation1)
    4.     .withOperator("uid2", transformation2)
    5.     .write(savepointPath);
    1. Savepoint
    2.     .create(backend, 128)
    3.     .withOperator("uid1", transformation1)
    4.     .withOperator("uid2", transformation2)
    5.     .write(savepointPath)

    Besides creating a savepoint from scratch, you can base on off an existing savepoint such as when bootstrapping a single new operator for an existing job.

    1. Savepoint
    2.     .load(backend, oldPath)
    3.     .withOperator("uid", transformation)
    4.     .write(newPath);
    1. Savepoint
    2.     .load(backend, oldPath)
    3.     .withOperator("uid", transformation)
    4.     .write(newPath)

    Note: When basing a new savepoint on existing state, the state processor api makes a shallow copy of the pointers to the existing operators. This means that both savepoints share state and one cannot be deleted without corrupting the other!