Shulou(Shulou.com)06/03 Report--

Author: Baoniu

This article, translated from a blog post on streaml.io: "Exactly once is NOT exactly the same", analyzes the "Exactly Once" feature that is often said in stream computing systems. The main point is that "accurate once" is not guaranteed to be exactly the same. The main contents are as follows:

Background

1.1. Up to one time (At-most-once)

1.2. At least once (At-least-once)

1.3. Precision once (Exactly-once)

Is "accurate once" really "accurate once"?

Comparison of distributed snapshots with at least one event delivery and deduplication

Conclusion

Referenc

At present, the most widely used stream computing systems in the market are Apache Storm,Apache Flink, Heron, Apache Kafka (Kafka Streams) and Apache Spark (Spark Streaming). One widely discussed feature about stream computing systems is the "exactly-once" semantics, which many systems claim to have supported. However, there are many misunderstandings and ambiguities about what is "exactly-once" and how to realize "exactly-once". Next, let's do some analysis.

I. background

Stream processing (sometimes called event handling) can be described simply as continuous processing of × × data or events. A flow or event processing application can be described more or less as a directed graph and is usually described as a directed acyclic graph (DAG). In such a graph, each edge represents a stream of data or events, each vertex represents an operator, and data or events from adjacent edges are processed using the logic defined in the program. There are two special types of vertices, commonly called sources and sinks. Sources reads external data / events into the application, while sinks typically collects the results generated by the application. The following figure is an example of a streaming application.

Cdn.xitu.io/2019/5/31/16b0d1bc03b3b729?w=1080&h=910&f=jpeg&s=49487 ">

A typical stream processing topology

A flow processing engine usually allows users to specify reliability patterns or processing semantics to indicate what guarantees it will provide for data processing throughout the application. These guarantees make sense because you will always encounter failures that may lead to data loss due to networks, machines, etc. Stream processing engines typically provide three data processing semantics for applications: up to once, at least once, and precisely once.

The following is a loose definition of these different processing semantics:

Up to one time (At-most-once)

This is essentially a "best-effort" approach. Ensure that data or events are processed by all operators in the application at most once. This means that if the data is lost before it is fully processed by the streaming application, no other retries or retransmissions will take place. The example in the following figure illustrates this situation.

At-most-once processing semantics

At least once (At-least-once)

All operators in the application guarantee that data or events are processed at least once. This usually means that if the event is lost before the streaming application is fully processed, the event will be replayed or retransmitted from the source. However, because events can be retransmitted, an event is sometimes handled multiple times, which is called at least once.

The example in the figure below depicts the situation where the first operator fails to handle the event at first, then succeeds in the retry, and then succeeds in the second retry, which is not necessary.

At-least-once processing semantics

Precision once (Exactly-once)

Even in the case of a variety of failures, all operators in the streaming application guarantee that the event will only be handled "precisely once". (there are also articles that translate Exactly-once as: exactly once, exactly once.)

Two popular mechanisms are commonly used to implement "precise once" processing semantics.

Distributed snapshot / stateful checkpoint

At least one event transfer and deduplication of duplicate data

The implementation of the "accurate once" distributed snapshot / state checkpoint method is inspired by the Chandy-Lamport distributed snapshot algorithm [1]. With this mechanism, all states of each operator in the streaming application are checkpoint periodically. If a failure occurs anywhere in the system, all states of each operator are rolled back to the latest globally consistent checkpoint point. During the rollback, all processing is paused. The source is also reset to the correct offset corresponding to the most recent checkpoint. The entire streaming application basically returns to its last consistent state, and then the program can restart from that state. The following figure describes the basics of this checkpoint mechanism.

Distributed snapshot

In the figure above, the streaming application works properly at T1 time and checkpoint is done. However, at time T2, the operator failed to process the input data. At this point, the state value of Syst4 has been saved to persistent memory, while the state value of Spati12 is saved in the memory of the operator. To fix this discrepancy, at time T3, the processor rolls back the state to Swatches 4 and "replays" each continuous state in the stream until recently, and processes each data. The end result is that some data has been processed many times, but that doesn't matter, because the resulting state is the same no matter how many rollbacks are performed.

Another way to achieve "precision once" is to achieve at least one event transfer on each operator and to deduplicate data. The flow processing engine using this method replays the failed event so that it further attempts to process and remove the repeated event for each operator before the event enters the user-defined logic in the operator. This mechanism requires that a transaction log be maintained for each operator to track the events it has handled. The engines that take advantage of this mechanism are Google's MillWheel [2] and Apache Kafka Streams. The following figure illustrates the main points of this mechanism.

At-least-once delivery plus deduplication

Second, is "accurate once" really "accurate once"?

Now let's re-examine the true assurance of "precise once" processing semantics to the end user. The term "precise once" can be misleading when it is described exactly once.

Some people may think that "precise once" describes the guarantee of event handling, where each event in the stream is handled only once. In fact, no engine can guarantee that it will only be processed once. In the face of any failure, it is impossible to guarantee that the user-defined logic in each operator will be executed only once in each event, because there is always the possibility that the user code will be partially executed.

Consider a scenario with a stream processing operator that performs a mapping operation to print the ID of the incoming event and then returns the event unchanged. The following pseudo-code illustrates this operation:

Map (Event event) {Print "Event ID:" + event.getId () Return event}

Each event has a GUID (globally unique ID). If the precise execution of the user logic is guaranteed once, then the event ID will be output only once. However, this is not guaranteed because failures can occur at any time during the execution of user-defined logic. The engine cannot determine on its own when the user-defined processing logic is executed. Therefore, there is no guarantee that any user-defined logic will be executed only once. This also means that external operations implemented in user-defined logic, such as writing to a database, are not guaranteed to be performed only once. Such operations still need to be performed in an idempotent manner.

So what do engines guarantee when they declare that semantics are handled "precisely once"? If there is no guarantee that user logic will be executed only once, what logic will be executed only once? When engines declare that "precise once" processing semantics, they are actually saying that they can guarantee that engine-managed state updates are submitted only once to persistent back-end storage.

Both of the mechanisms described above use persistent back-end storage as a source of authenticity, saving the state of each operator and automatically submitting updates to it. For mechanism 1 (distributed snapshot / stateful checkpoint), this persistent back-end state is used to hold the globally consistent state checkpoint (checkpoint state of each operator) of the streaming application. For mechanism 2 (at least one event delivery plus deduplication), the persistent back-end state is used to store the state of each operator and the transaction log of each operator, which tracks all events that it has fully processed.

The submission status or the application of updates to the persistent backend as the real source can be described as happening exactly once. However, as mentioned above, an update / change in the calculation state, that is, an event that executes arbitrary user-defined logic on an event, may occur more than once if a failure occurs. In other words, the handling of an event can occur multiple times, but the effect of this processing is reflected only once in the persistent back-end state store. Therefore, we believe that the best term to effectively describe the semantics of these processes is "effectively once".

So what do engines guarantee when they declare that semantics are handled "precisely once"? If there is no guarantee that user logic will be executed only once, what logic will be executed only once? When engines declare that "precise once" processing semantics, they are actually saying that they can guarantee that engine-managed state updates are submitted only once to persistent back-end storage.

III. Comparison of distributed snapshots with at least one event delivery and deduplication

From a semantic point of view, distributed snapshots provide the same guarantee as at least one event delivery and deduplication mechanism. However due to the implementation differences between the two mechanisms there are significant performance differences.

The performance overhead of mechanism 1 (distributed snapshot / stateful checkpoint) is minimal, because the engine actually sends regular and special events together to all operators in the streaming application, while stateful checkpoints can be performed asynchronously in the background. However, for large streaming applications, failures may occur more frequently, causing the engine to pause the application and roll back the state of all operators, which in turn affects performance. The larger the streaming application, the more likely it is to fail, so the more frequent it is, and in turn, the greater the impact on the performance of the streaming application. However, this mechanism is non-intrusive and the additional resources required by the runtime have little impact.

Mechanism 2 (at least one event delivery plus deduplication) may require more resources, especially storage *. Using this mechanism, the engine needs to be able to track each tuple that each operator instance has fully processed to perform deduplication, as well as deduplication itself for each event. This means that a large amount of data needs to be tracked, especially if the streaming application is large or there are many applications running. Each event on each operator that performs deduplication has a performance overhead. However, with this mechanism, the performance of streaming applications is unlikely to be affected by the size of the application. For mechanism 1, if any operator fails, global pause and state rollback need to occur; for mechanism 2, the impact of failure is more local. When a failure occurs in the operator, events that may not have been fully handled are replayed / retransmitted only from the upstream source. The performance impact is isolated from the location of the failure in the streaming application, and there is little impact on the performance of other operators in the streaming application. From a performance perspective, the advantages and disadvantages of these two mechanisms are as follows.

Advantages and disadvantages of distributed snapshots / stateful checkpoints:

Advantages:

Less performance and resource overhead

Disadvantages:

It has a great impact on performance.

The larger the topology, the greater the potential impact on performance

Advantages and disadvantages of at least one event delivery and deduplication mechanism:

Advantages:

The impact of failure on performance is local

The impact of the failure does not necessarily increase with the size of the topology

Disadvantages:

May require a large amount of storage and infrastructure to support

Performance cost per event per operator

Although there is a theoretical difference between distributed snapshots and at least one event transfer plus deduplication mechanism, both can be reduced to at least one processing plus idempotency. For both mechanisms, when a failure occurs (at least once), the event is replayed / retransmitted, and through state rollback or event deduplication, the operator is essentially idempotent in updating the internal management state.

IV. Conclusion

In this blog post, I hope to convince you that the word "precise once" is very misleading. Providing "precise once" processing semantics actually means that different updates to the operator state managed by the flow processing engine are reflected only once. "precise once" does not guarantee the handling of events, that is, the execution of any user-defined logic will occur only once. We prefer to use the term "effectively once" to describe this guarantee, because processing does not necessarily guarantee that it only happens once, but the impact on the state of engine management is reflected only once. Two popular mechanisms, distributed snapshot and deduplication, are used to implement precise / effective one-time processing semantics. These two mechanisms provide the same semantic guarantee for message processing and state update, but there are differences in performance. This article is not intended to convince you that any mechanism is superior to the other, because they have their own advantages and disadvantages.

V. reference

Chandy, K. Mani and Leslie Lamport.Distributed snapshots: Determining global states of distributed systems. ACMTransactions on Computer Systems (TOCS) 3. 1 (1985): 63-75.

Akidau, Tyler, et al. MillWheel:Fault-tolerant stream processing at internet scale. Proceedings of the VLDBEndowment 6.11 (2013): 1033-1044.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.