Talk about the state management mechanism in Flink framework

Status overview

In all current scenarios of streaming Computing , Divide the state of the data flow into

A stateful

and

No state

Two types of .

Stateless means that stateless computation observes each independent event , And only output the result according to the last event

. Take a chestnut ： A stream handler , The temperature data is received from the sensor and then measured at a temperature of 90 The alarm message is sent out in centigrade .

Stateful calculations output results based on multiple events

. Take a chestnut ： Calculate the average temperature for the past hour , It's stateful computation 、 If two differences are received within one minute  20  Temperature readings above degrees , Then issue a warning, etc .

As can be seen from the above figure ：

• Stateless stream processing receives each data record separately , Then the output data is generated according to the latest input data .（ Only one input record is converted at a time , And only record the output according to the latest input ）

• Stateful flow processing maintains state , The output record is generated based on the latest input record and the current status value .（ Maintain status values for all processed records , And update the status according to each newly entered record , Therefore, the output records reflect the results after considering multiple events .）

Flink The state of

Flink There is a task to maintain the status in , And all the data used to calculate a result , All belong to the state of this task . In most cases we can put Flink The state in is understood as a local variable , Stored in memory . States are always associated with specific operators , stay flink Status registration is required in .

<center>（ This is from the Internet ）

Flink There are two types of states in the framework ： Operator state 、 Keying state . Let's talk about these two states in detail .

Operator state

The scope of the operator state is limited to the operator task . All data processed by the same parallel task can access the same state . in other words , Shared in the same task . Be careful ：

Operator state cannot be accessed by another subtask of the same or different operator

<center>（ This is from the Internet ）

Flink  Three basic data structures are provided for operator states ：

List status

• Represent a state as a list of data .

Union List status

• State is also represented as a list of data . It differs from the regular list state in that , In case of failure , Or from the repository （savepoint） How to recover when starting an application .

Broadcast status

• If an operator has multiple tasks , And each task has the same status , So this particular situation is best suited to broadcast state

The code is as follows ：

public class StateTest1_OperatorState {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 // Facility parallelism is 1
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);
 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });

 //  Define a stateful Map operation   Count the number of current partition data
 SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMap());

 resultStream.print();
 env.execute();
 }

 // Customize mapFunction  Registration status implementation ListCheckpointed
 private static class MyCountMap implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer>{

 //  Define a local variable as the operator state
 private Integer count = 0;

 @Override
 public Integer map(SensorReading sensorReading) throws Exception {
 count++;
 return count;
 }

 @Override
 public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
 return Collections.singletonList(count);
 }

 @Override
 public void restoreState(List<Integer> state) throws Exception {
 for (Integer number : state) {
 count+=number;
 }
 }
 }
}


Keying state

The keying state is based on the key defined in the input data stream （key） To maintain and access .Flink  For each  key  Maintain a status instance , And will have all the data with the same key , They are all partitioned into the same operator task , This task will maintain and handle this  key  Corresponding state . When a task processes a piece of data , Its current state is automatically limited to  key.

<center>（ This is from the Internet ）

Flink  Three basic data structures are provided for keying states ：

Value status

• Represent the state as a single value .

List status

• Represent a state as a list of data

Map state

• Represent the status as a set of  Key-Value  Yes

Aggregate state （Reducing state & Aggregating State）

• Represent the state as a list for aggregation operations

The code is as follows ：

public class StateTest2_KeyedState {
 public static void main(String[] args) throws Exception{
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 // Facility parallelism is 1
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);
 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });

 //  Define a stateful Map operation   Count the present senser The number of data
 SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy(&quot;id&quot;)
 .map(new MyKeyCouneMap());

 resultStream.print();
 env.execute();
 }

 //  Customize RichMapFunction
 private static class MyKeyCouneMap extends RichMapFunction<SensorReading,Integer> {
 //  Declare keying state
 private ValueState<Integer> KeyCouneState ;

 //private ListState<String> listState;

 //private MapState<String,Double> mapState;

 // private ReducingState<SensorReading> reduceState;

 @Override
 public void open(Configuration parameters) throws Exception {
 KeyCouneState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>(
 &quot;key-count&quot;,Integer.class
 ));

 // listState =getRuntimeContext().getListState(new ListStateDescriptor<String>(
 // &quot;list-count&quot;,String.class
 ));

 //mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Double>(
 // &quot;map-count&quot;,String.class,Double.class
 ));

 //reduceState =getRuntimeContext().getReducingState(new ReducingStateDescriptor<SensorReading>(
 // &quot;reducing-count&quot;,SensorReading.class
 //));class
 }

 @Override
 public Integer map(SensorReading sensorReading) throws Exception {
 //  Read status
 Integer count = KeyCouneState.value();
 if (count==null){
 count = 0;
 }else {
 count++;
 }
 //  Update status , Assign a value to the state
 KeyCouneState.update(count);
 return count;

 // listState
 //Iterable<String> iterable = listState.get();
 //for (String s : listState.get()) {
 // System.out.println(s);
 //}
 //listState.add(&quot;hello&quot;);

 // mapState
 //mapState.get(&quot;1&quot;);
 //mapState.put(&quot;2&quot;,12.3);

 }
 }
}


State backend

State storage 、 Access and maintenance , Determined by a pluggable component , This component is called the state backend . The state backend is mainly responsible for two things ： Local state management , And will checkpoint （checkpoint） Status is written to remote storage . There are three types of state backend ：

MemoryStateBackend

• Memory level state backend , The keying state is managed as an object in memory , Store them in TaskManager  Of  JVM  Pile it up , And will be  checkpoint  Stored in JobManager  The memory of the . characteristic ： Fast 、 Low latency , But not stable .

FsStateBackend

• take  checkpoint  To a remote persistent file system （FileSystem） On , And for the local state , Follow  MemoryStateBackend  equally , There are also  TaskManager  Of  JVM  Memory level local access speed is also available on the heap , And better fault tolerance .

RocksDBStateBackend

• After serializing all States , Deposited locally  RocksDB  Storage in

Configure the status backend in the code ：

public class StateTest4_FaultTolerance {
 public static void main(String[] args) throws Exception{
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

 //  Status backend configuration
 env.setStateBackend(new MemoryStateBackend());
 env.setStateBackend(new FsStateBackend(""));
 
 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);
 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 env.execute();
 }
}


flink Configuration in the configuration file ：

A case ：

Check the temperature jump of industrial IOT sensor , If the difference between two consecutive temperatures exceeds 10 degree , Just call the police .

The code is as follows

public class StateTest3_ApplicationCase {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 // Facility parallelism is 1
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);
 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });

 //  Define a flutmap operation , Detect temperature jump output alarm
 SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream.keyBy(&quot;id&quot;)
 .flatMap(new TempChangeWarring(10.0));

 resultStream.print();
 env.execute();
 }

 private static class TempChangeWarring extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>> {

 //  Set the temperature jump threshold
 private Double threshold;

 public TempChangeWarring(Double threshold) {
 this.threshold = threshold;
 }

 //  Define the State , Save the last temperature value
 private ValueState<Double> lastTempState;

 @Override
 public void open(Configuration parameters) throws Exception {
 lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>(
 &quot;lastTemp&quot;, Double.class
 ));
 }

 @Override
 public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception {
 //  To obtain state
 Double lastTemp = lastTempState.value();
 //  If not for null  Just calculate the temperature difference twice
 if (lastTemp != null) {
 Double diff = Math.abs(value.getTemperature() - lastTemp);
 if (diff>=threshold){
 out.collect(new Tuple3<>(value.getId(),lastTemp,value.getTemperature()));
 }
 }

 //  Update status
 lastTempState.update(value.getTemperature());
 }

 @Override
 public void close() throws Exception {
 lastTempState.clear();
 }
 }
}