Asynchronous in-memory data checkpointing for distributed computing systems
View Patent ↗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:
maintaining, by a checkpoint manager system, global checkpoint metadata associated with asynchronous checkpoint operations that are independently performed by a plurality of operators to generate checkpoints of operator states while processing a stream of data records;
receiving, by the checkpoint manager system, a checkpoint acknowledgment message from a given operator of the plurality of operators, wherein the checkpoint acknowledgment message comprises metadata which indicates a completion of an asynchronous checkpoint operation performed by the given operator to generate a checkpoint of the operator state that is associated with a given block of data records in the stream of data records;
updating, by the checkpoint manager system, the global checkpoint metadata based on the metadata included in the checkpoint acknowledgment message received from the given operator; and
utilizing, by the checkpoint manager system, the global checkpoint metadata to track a status of the asynchronous checkpointing operations performed by the plurality of operators with respect to the given block of data records.
2. The method of claim 1 , wherein the metadata of the checkpoint acknowledgment message comprises a unique block identifier that identifies the given block of data records associated with the checkpoint, and a unique checkpoint identifier of the checkpoint that is generated by the given operator for the given block of data records.
3. The method of claim 1 , wherein utilizing the global checkpoint metadata to track the status of the asynchronous checkpointing operations performed by the plurality of operators with respect to the given block of data records comprises:
utilizing, by the checkpoint manager system, the global checkpoint metadata to determine a completion of the asynchronous checkpointing operations by the plurality of operators with respect to the given block of data records; and
updating a status of the asynchronous checkpointing operations for the given block of data records as being complete, when the plurality of operators have completed their asynchronous checkpoint operations with respect to the given block of data records to thereby generate a distributed checkpoint comprising all checkpoints of operators states of the plurality of operators for the given data block of records.
4. The method of claim 3 , further comprising:
committing, by the checkpoint manager system, the distributed checkpoint associated with the given data block of records; and
sending, by the checkpoint manager system, a notification to the plurality of operators to purge from memory the checkpoints of operator states and data records associated with the given data block of records.
5. The method of claim 4 , further comprising:
receiving, by the checkpoint manager system, a request from a checkpoint handler to access the global checkpoint metadata information and determine a previous committed distributed checkpoint; and
providing, by the checkpoint manager system, information to the checkpoint handler regarding the previous committed distributed checkpoint, which allows the checkpoint handler to access from persistent storage the checkpoints of operator states associated with the previous committed distributed checkpoint.
6. The method of claim 1 , wherein the global checkpoint metadata maintained by the checkpoint manager system comprises (i) a topology identifier that uniquely identifies a topology of the plurality of operators that are configured to process the stream of data records, (ii) a stream identifier that uniquely identifies the stream of data records being processed by the plurality of operators within the topology, and (iii) block identifiers that uniquely identify different blocks of data records in the stream of data records for which checkpoints of operator states are generated by the plurality of operators.
7. The method of claim 6 , wherein the global checkpoint metadata maintained by the checkpoint manager system further comprises a unique checkpoint identifier associated with a given checkpoint of an operator state for a given block of data records with a given block identifier.
8. The method of claim 6 , wherein the global checkpoint metadata maintained by the checkpoint manager further comprises status metadata which indicates a checkpoint status, wherein the checkpoint status is specified as one of an initial state, an in-progress state, and a completed state.
9. 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:
maintaining, by a checkpoint manager system, global checkpoint metadata associated with asynchronous checkpoint operations that are independently performed by a plurality of operators to generate checkpoints of operator states while processing a stream of data records;
receiving, by the checkpoint manager system, a checkpoint acknowledgment message from a given operator of the plurality of operators, wherein the checkpoint acknowledgment message comprises metadata which indicates a completion of an asynchronous checkpoint operation performed by the given operator to generate a checkpoint of the operator state that is associated with a given block of data records in the stream of data records;
updating, by the checkpoint manager system, the global checkpoint metadata based on the metadata included in the checkpoint acknowledgment message received from the given operator; and
utilizing, by the checkpoint manager system, the global checkpoint metadata to track a status of the asynchronous checkpointing operations performed by the plurality of operators with respect to the given block of data records.
10. The article of manufacture of claim 9 , wherein the metadata of the checkpoint acknowledgment message comprises a unique block identifier that identifies the given block of data records associated with the checkpoint, and a unique checkpoint identifier of the checkpoint that is generated by the given operator for the given block of data records.
11. The article of manufacture of claim 10 , wherein the program code for utilizing the global checkpoint metadata to track the status of the asynchronous checkpointing operations performed by the plurality of operators with respect to the given block of data records comprises program code that is executable by the processor to perform a method which comprises:
utilizing, by the checkpoint manager system, the global checkpoint metadata to determine a completion of the asynchronous checkpointing operations by the plurality of operators with respect to the given block of data records; and
updating a status of the asynchronous checkpointing operations for the given block of data records as being complete, when the plurality of operators have completed their asynchronous checkpoint operations with respect to the given block of data records to thereby generate a distributed checkpoint comprising all checkpoints of operators states of the plurality of operators for the given data block of records.
12. The article of manufacture of claim 11 , further comprising program code that is executable by the processor to perform a method which comprises:
committing, by the checkpoint manager system, the distributed checkpoint associated with the given data block of records; and
sending, by the checkpoint manager system, a notification to the plurality of operators to purge from memory the checkpoints of operator states and data records associated with the given data block of records.
13. The article of manufacture of claim 12 , further comprising program code that is executable by the processor to perform a method which comprises:
receiving, by the checkpoint manager system, a request from a checkpoint handler to access the global checkpoint metadata information and determine a previous committed distributed checkpoint; and
providing, by the checkpoint manager system, information to the checkpoint handler regarding the previous committed distributed checkpoint, which allows the checkpoint handler to access from persistent storage the checkpoints of operator states associated with the previous committed distributed checkpoint.
14. The article of manufacture of claim 9 , wherein the global checkpoint metadata maintained by the checkpoint manager system comprises (i) a topology identifier that uniquely identifies a topology of the plurality of operators that are configured to process the stream of data records, (ii) a stream identifier that uniquely identifies the stream of data records being processed by the plurality of operators within the topology, and (iii) block identifiers that uniquely identify different blocks of data records in the stream of data records for which checkpoints of operator states are generated by the plurality of operators.
15. The article of manufacture of claim 14 , wherein the global checkpoint metadata maintained by the checkpoint manager system further comprises a unique checkpoint identifier associated with a given checkpoint of an operator state for a given block of data records with a given block identifier.
16. The article of manufacture of claim 14 , wherein the global checkpoint metadata maintained by the checkpoint manager further comprises status metadata which indicates a checkpoint status, wherein the checkpoint status is specified as one of an initial state, an in-progress state, and a completed state.
17. A server 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 implement a checkpoint manager system, wherein the checkpoint manager system is configured to:
maintain global checkpoint metadata associated with asynchronous checkpoint operations that are independently performed by a plurality of operators to generate checkpoints of operator states while processing a stream of data records;
receive a checkpoint acknowledgment message from a given operator of the plurality of operators, wherein the checkpoint acknowledgment message comprises metadata which indicates a completion of an asynchronous checkpoint operation performed by the given operator to generate a checkpoint of the operator state that is associated with a given block of data records in the stream of data records;
update the global checkpoint metadata based on the metadata included in the checkpoint acknowledgment message received from the given operator; and
utilize the global checkpoint metadata to track a status of the asynchronous checkpointing operations performed by the plurality of operators with respect to the given block of data records.
18. The server node of claim 17 , wherein the checkpoint manager system is configured to:
utilize the global checkpoint metadata to determine a completion of the asynchronous checkpointing operations by the plurality of operators with respect to the given block of data records; and
update a status of the asynchronous checkpointing operations for the given block of data records as being complete, when the plurality of operators have completed their asynchronous checkpoint operations with respect to the given block of data records to thereby generate a distributed checkpoint comprising all checkpoints of operators states of the plurality of operators for the given data block of records.
19. The server node of claim 18 , wherein the checkpoint manager system is configured to:
commit the distributed checkpoint associated with the given data block of records; and
send a notification to the plurality of operators to purge from memory the checkpoints of operator states and data records associated with the given data block of records.
20. The server node of claim 19 , wherein the checkpoint manager system is configured to:
receive a request from a checkpoint handler to access the global checkpoint metadata information and determine a previous committed distributed checkpoint; and
provide information to the checkpoint handler regarding the previous committed distributed checkpoint, which allows the checkpoint handler to access from persistent storage the checkpoints of operator states associated with the previous committed distributed checkpoint.