NOTE: This document is a design draft and does not reflect the state of a stable release, but the work in progress on the current master, and future considerations.

This document is intended to give an overview of how operator state is handled and represented in streaming programs.

State access patterns in streaming programs

To design a properly functioning, scalable state representation we need to understand the different state access and handling patterns that applications need in streaming programs. We can distinguish three main state types: local state, partitioned state, out-of-core state. 

Local (non-partitioned) state

Local or non-partitioned state is the simplest form of operator state which represent the current state of a specific operator instance in a parallel streaming operator. Local states stored at different operator instances do not interact with each other. For instance if we have a mapper with parallelism of 10 that means each parallel instance holds it’s own local state. An important thing to note here is that state updates reflected in a particular local state will depend only on the input of that specific operator instance. As there is no strict control over the input which each operator instance processes (only via stream partitioning), the operator itself needs to deal with this non-deterministic nature of the inputs and this also limits expressivity. 

 Typical usage of non-partitioned state includes source operators (storing offsets), or any global aggregation/summary or analysis operators where the local results will be merged into a global result afterwards.

Scaling out non-partitioned state can in most cases be done by just starting a new operator instance with a blank state, or a user supplied splitting function can be used to split the state of an existing instance. For reducing job parallelism the user should  generally provide a function that can merge 2 local states to maintain correct results.

val env: StreamExecutionEnvironment = ...
 
val stream = env.addSource(new SourceFunction[Int]() {
 	
	// store current offset as operator state
    var offset: OperatorState[Int] = _
	var isRunning = false
 
    override def open(config: Configuration): Unit = {
		// get non-partitioned state from context with default value of 0
        offset = getRuntimeContext().getState("offset", 0, false)
    }
 
    override def run(ctx: SourceContext[Int]): Int = {
		isRunning = true
        while (isRunning) {
			// output the current offset then increment
			val currOffset = offset.value
			ctx.collect(currOffset)
			offset.update(currOffset + 1)
		}
    }
 
	override def cancel(): Unit = { isRunning = false }
})

class CheckpointedSource extends SourceFunction[Int] with Checkpointed[Int] {
    var offset: Int
	var isRunning = false
 
    override def run(ctx: SourceContext[Int]): Int = {
		isRunning = true
        while (isRunning) {
			ctx.collect(offset)
			offset += 1
		}
    }
 
	override def snapshotState(checkpointID: Long, timeStamp: Long): Int = offset
	override def restoreState(state: Int): Unit = {offset = state}
 
	override def cancel(): Unit = { isRunning = false }
})

val stream: DataStream[String] = ...
val keyed: KeyedDataStream[String] = stream.keyBy(x => x)
 
val  = keyed.map(new RichMapFunction[String, (String, Int)]() {
 
    val countPerID: OperatorState[Int] = _
 
    override def open(config: Configuration): Unit = {
		//the true flag marks the state as partitioned
        countPerID = getRuntimeContext().getState("count", 0, true)
    }
 
    override def map(in: String): (String, Int) = {
		// countPerID holds the state for the current input
        val countSoFar= countPerID.value + 1
        countPerID.update(countSoFar)
        (in, countSoFar)
    }
})