Asynchronous in-memory data checkpointing for distributed computing systems
Techniques are provided for implementing asynchronous checkpointing of in-memory data in a distributed computing system. For example, a method includes processing a stream of data records by an operator executing on a computing node, maintaining in a system memory, an operator state which is generated in response to the operator processing the stream of data records, and performing an asynchronous checkpointing process. The asynchronous checkpointing process includes enqueuing a checkpoint of the operator state in a first queue, wherein the first queue is maintained in the system memory, and executing a background worker thread to dequeue the checkpoint of the operator state from the first queue and store the checkpoint of the operator state in a data store. The operator continues with processing the stream of data records during the asynchronous checkpointing process.
1. A method, comprising:
processing a stream of data records by an operator executing on a computing node;
maintaining in a system memory of the computing node, an operator state which is generated in response to the operator processing the stream of data records; and
performing an asynchronous checkpointing process by a checkpointing system executing on the computing node, wherein the asynchronous checkpointing process comprises:
enqueuing a checkpoint of the operator state in a first queue, wherein the first queue is maintained in the system memory of the computing node;
executing a background worker thread on the computing node to dequeue the checkpoint of the operator state from the first queue and store the checkpoint of the operator state in a data store;
sending metadata to a central checkpoint manager, wherein the metadata comprises information which is configured to enable the central checkpoint manager to track a status of the asynchronous checkpointing process;
wherein the asynchronous checkpointing process is performed independent of asynchronous checkpointing operations performed by one or more other operators processing the stream of data records; and
wherein the operator continues with processing the stream of data records during the asynchronous checkpointing process.
2. The method of claim 1 , wherein enqueuing the checkpoint of the operator state in the first queue comprises:
serializing the operator state, and
storing a checkpoint of the serialized operator state in the first queue.
3. The method of claim 1 , wherein the asynchronous checkpointing process further comprises:
dequeuing, by the background worker thread, a plurality of checkpoints of operator states of different operators, which are stored in the first queue;
batch processing the plurality of checkpoints dequeued from the first queue to store the plurality of checkpoints in the data store.
4. The method of claim 3 , wherein the asynchronous checkpointing process further comprises compressing the plurality of checkpoints of the operator states, by the background worker thread, and storing compressed checkpoints of the operator states in the data store.
5. The method of claim 1 , wherein the first queue in the system memory comprises a first-in-first-out (FIFO) queue structure.
6. The method of claim 1 , wherein enqueuing the checkpoint of the operator state in the first queue comprises embedding metadata with the checkpoint of the operator state, wherein the metadata comprises a unique checkpoint identifier of the checkpoint, and a window identifier which identifies a block of data records within the stream of data records which corresponds to the checkpoint.
7. The method of claim 1 , wherein the asynchronous checkpointing process further comprises enqueuing, by the background worker thread, a checkpoint acknowledgement record in a second queue which resides in the system memory, after the checkpoint of the operator state is stored in the data store, wherein the checkpoint acknowledgement record indicates that the storage of the checkpoint of the operator state in the data store is completed.
8. The method of claim 7 , wherein the asynchronous checkpointing process further comprises:
dequeuing, by the background worker thread, the checkpoint acknowledgement record from the second queue; and
sending the acknowledgment record to the central checkpoint manager.
9. The method of claim 8 , wherein the metadata that is sent to the central checkpoint manager comprises metadata contained in the checkpoint acknowledgement record, wherein the metadata comprises a unique checkpoint identifier of the checkpoint, and a window identifier which identifies a block of data records within the stream of data records which corresponds to the checkpoint;
wherein the central checkpoint manager maintains global checkpoint metadata associated with a topology of operators that are configured to process the stream of data records;
wherein the global metadata is utilized to track a completion of a checkpoint operation by all operators within the topology of operators with respect to checkpoints of the operator states of all the operators which have a same window identifier.
10. An article of manufacture comprising a processor-readable storage medium having stored program code which is executable by a processor to implement a method which comprises:
processing a stream of data records by an operator executing on a computing node;
maintaining in a system memory of the computing node, an operator state which is generated in response to the operator processing the stream of data records; and
performing an asynchronous checkpointing process by a checkpointing system executing on the computing node, wherein the asynchronous checkpointing process comprises:
enqueuing a checkpoint of the operator state in a first queue, wherein the first queue is maintained in the system memory of the computing node;
executing a background worker thread on the computing node to dequeue the checkpoint of the operator state from the first queue and store the checkpoint of the operator state in a data store;
sending metadata to a central checkpoint manager, wherein the metadata comprises information which is configured to enable the central checkpoint manager to track a status of the asynchronous checkpointing process;
wherein the asynchronous checkpointing process is performed independent of asynchronous checkpointing operations performed by one or more other operators processing the stream of data records; and
wherein the operator continues with processing the stream of data records during the asynchronous checkpointing process.
11. The article of manufacture of claim 10 , wherein enqueuing the checkpoint of the operator state in the first queue comprises:
serializing the operator state, and
storing a checkpoint of the serialized operator state in the first queue.
12. The article of manufacture of claim 10 , wherein the program code that is executable by the processor to perform the asynchronous checkpointing process further comprises program code that is executable by the processor to perform a method which comprises:
dequeuing, by the background worker thread, a plurality of checkpoints of operator states of different operators, which are stored in the first queue;
batch processing the plurality of checkpoints dequeued from the first queue to store the plurality of checkpoints in the data store.
13. The article of manufacture of claim 12 , wherein the program code that is executable by the processor to perform the asynchronous checkpointing process further comprises program code that is executable by the processor to perform a method which comprises compressing the plurality of checkpoints of the operator states, by the background worker thread, and storing compressed checkpoints of the operator states in the data store.
14. The article of manufacture of claim 10 , wherein the first queue in the system memory comprises a first-in-first-out (FIFO) queue structure.
15. The article of manufacture of claim 10 , wherein enqueuing the checkpoint of the operator state in the first queue comprises embedding metadata with the checkpoint of the operator state, wherein the metadata comprises a unique checkpoint identifier of the checkpoint, and a window identifier which identifies a block of data records within the stream of data records which corresponds to the checkpoint.
16. The article of manufacture of claim 10 , wherein the program code that is executable by the processor to perform the asynchronous checkpointing process further comprises program code that is executable by the processor to perform a method which comprises enqueuing, by the background worker thread, a checkpoint acknowledgement record in a second queue which resides in the system memory, after the checkpoint of the operator state is stored in the data store, wherein the checkpoint acknowledgement record indicates that the storage of the checkpoint of the operator state in the data store is completed.
17. The article of manufacture of claim 16 , wherein the program code that is executable by the processor to perform the asynchronous checkpointing process further comprises program code that is executable by the processor to perform a method which further comprises:
dequeuing, by the background worker thread, the checkpoint acknowledgement record from the second queue; and
sending the acknowledgment record to the central checkpoint manager.
18. The article of manufacture of claim 17 , wherein the metadata that is sent to the central checkpoint manager comprises metadata contained in the checkpoint acknowledgement record, wherein the metadata comprises a unique checkpoint identifier of the checkpoint, and a window identifier which identifies a block of data records within the stream of data records which corresponds to the checkpoint;
wherein the central checkpoint manager maintains global checkpoint metadata associated with a topology of operators that are configured to process the stream of data records;
wherein the global metadata is utilized to track a completion of a checkpoint operation by all operators within the topology of operators with respect to checkpoints of the operator states of all the operators which have a same window identifier.
19. A computing node comprising:
a system memory configured to store program instructions; and
at least one processor configured to execute the stored program instructions in the system memory to perform a method comprising:
processing a stream of data records by an operator executing on a computing node;
maintaining in a system memory of the computing node, an operator state which is generated in response to the operator processing the stream of data records; and
performing an asynchronous checkpointing process by a checkpointing system executing on the computing node, wherein the asynchronous checkpointing process comprises:
enqueuing a checkpoint of the operator state in a first queue, wherein the first queue is maintained in the system memory of the computing node;
executing a background worker thread on the computing node to dequeue the checkpoint of the operator state from the first queue and store the checkpoint of the operator state in a data store;
sending metadata to a central checkpoint manager, wherein the metadata comprises information which is configured to enable the central checkpoint manager to track a status of the asynchronous checkpointing process;
wherein the asynchronous checkpointing process is performed independent of asynchronous checkpointing operations performed by one or more other operators processing the stream of data records; and
wherein the operator continues with processing the stream of data records during the asynchronous checkpointing process.
20. The computing node of claim 19 , wherein the asynchronous checkpointing process further comprises:
enqueuing, by the background worker thread, a checkpoint acknowledgement record in a second queue which resides in the system memory, after the checkpoint of the operator state is stored in the data store, wherein the checkpoint acknowledgement record indicates that the storage of the checkpoint of the operator state in the data store is completed;
dequeuing, by the background worker thread, the checkpoint acknowledgement record from the second queue; and
sending the acknowledgment record to a central checkpoint manager.