Talk about the understanding of fault tolerance mechanism and state consistency in Flink framework

Fault tolerance mechanism

Flink  The core of fault recovery mechanism , Is the consistency checkpoint of the application state , Consistent checkpoints for stateful flow applications , It's actually the state of all tasks , A copy at some point in time （ A snapshot ）; At this point in time , It should be the time when all tasks have just finished processing the same input data . During execution of the streaming application ,Flink  Consistent checkpoints that periodically save status , If something goes wrong , Flink  The most recent checkpoint will be used to consistently restore the state of the application , and . Restart the process .

Recovery steps ：

• restart app

• from  checkpoint  Read status in , Reset the status

• Start consuming and processing all data from checkpoint to failure

Flink  It also provides a user-defined image saving function , Just save the point ,Flink Savepoints are not automatically created , So the user （ Or an external scheduler ） The create operation must be explicitly triggered , Savepoints are a powerful feature . In addition to failure recovery , Savepoints can be used for ： Scheduled manual backups , Update the application , Version migration , Pause and restart the application .

Checkpoint configuration code case ：

public class Test{
 public static void main(String[] args) throws Exception{
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 //  Configuration of checkpoints
 //  The checkpoint start cycle is milliseconds
 env.enableCheckpointing(300);
 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 //  Set timeout
 env.getCheckpointConfig().setCheckpointTimeout(60000);
 //  Set the maximum execution checkp Number
 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
 //  Set the minimum interval , Two checkpoint Between , The interval between the previous completion and the next start cannot be less than XXX
 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100);
 //env.getCheckpointConfig().setPreferCheckpointForRecovery();
 //  tolerate checkpoint How many times have you failed
 env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);

 //  Configuration of restart policy
 //  Fixed delay restart   every other 10 Seconds to restart three times
 env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,10000L));
 //  Failure rate restart   stay 10 Count restarts within minutes , stay 10 Try to restart three times in minutes , Every septum 1 Minutes to restart
 env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10),Time.minutes(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]));
 });
 
 env.execute();
 }
}


State consistency

Stateful streaming , Each internal operator task can have its own state , Inside the stream processor , The so-called state consistency , In fact, we say that the calculation results should be accurate , A piece of data should not be lost , Nor should double counting , In case of failure, the state can be restored , Recalculate later , The result should also be completely correct .

State consistency classification ：

• AT-MOST-ONCE（ At most once ） When the mission fails , The easiest way is to do nothing , Neither restore the lost state , No replay of lost data .At-most-once  The meaning of semantics is to handle an event at most once .

• AT-LEAST-ONCE（ At least once ） In most real application scenarios , We hope not to lose Events . This type of protection is called  atleast-once, It means that all the events have been dealt with , And some events can be handled multiple times .

• EXACTLY-ONCE（ Exactly once ） Just once is the strictest guarantee , It's also the most difficult . Just dealing with semantics once doesn't just mean that no events are lost , It also means for every data , The internal state is only updated once .

Flink The source code provides an enumeration class for related configuration ：

@Public
public enum CheckpointingMode {

 /**
 * Sets the checkpointing mode to "exactly once". This mode means that the system will
 * checkpoint the operator and user function state in such a way that, upon recovery,
 * every record will be reflected exactly once in the operator state.
 *
 * <p>For example, if a user function counts the number of elements in a stream,
 * this number will consistently be equal to the number of actual elements in the stream,
 * regardless of failures and recovery.</p>
 *
 * <p>Note that this does not mean that each record flows through the streaming data flow
 * only once. It means that upon recovery, the state of operators/functions is restored such
 * that the resumed data streams pick up exactly at after the last modification to the state.</p>
 *
 * <p>Note that this mode does not guarantee exactly-once behavior in the interaction with
 * external systems (only state in Flink's operators and user functions). The reason for that
 * is that a certain level of &quot;collaboration&quot; is required between two systems to achieve
 * exactly-once guarantees. However, for certain systems, connectors can be written that facilitate
 * this collaboration.</p>
 *
 * <p>This mode sustains high throughput. Depending on the data flow graph and operations,
 * this mode may increase the record latency, because operators need to align their input
 * streams, in order to create a consistent snapshot point. The latency increase for simple
 * dataflows (no repartitioning) is negligible. For simple dataflows with repartitioning, the average
 * latency remains small, but the slowest records typically have an increased latency.</p>
 */
 EXACTLY_ONCE,

 /**
 * Sets the checkpointing mode to &quot;at least once&quot;. This mode means that the system will
 * checkpoint the operator and user function state in a simpler way. Upon failure and recovery,
 * some records may be reflected multiple times in the operator state.
 *
 * <p>For example, if a user function counts the number of elements in a stream,
 * this number will equal to, or larger, than the actual number of elements in the stream,
 * in the presence of failure and recovery.</p>
 *
 * <p>This mode has minimal impact on latency and may be preferable in very-low latency
 * scenarios, where a sustained very-low latency (such as few milliseconds) is needed,
 * and where occasional duplicate messages (on recovery) do not matter.</p>
 */
 AT_LEAST_ONCE
}


End to end state consistency

At present, the consistency guarantee we are in contact with is Flink  Stream processors are guaranteed internally ; And in real applications , Stream processing applications include data sources in addition to stream processors （ for example  Kafka） And output to the persistence system .

• flink Internal assurance ： rely on  checkpoint

• source  End ： An external source is required to reset the read location of the data

• sink  End ： Need to ensure recovery from failure , Data is not repeatedly written to external systems . Two ways ： Idempotent writing 、 Transaction write .

Different  Source  and  Sink  Consistency assurance of

（ This is from the Internet ）