Checkpoints in batch file processing
Embodiments of the present disclosure may provide a dynamic query execution model with fault tolerance and failure recovery techniques. Embodiments of the present disclosure may utilize checkpoints to map processed output files to their corresponding input files. Therefore, if an error occurs in processing one or more files, the system may only need to reschedule processing of selected file(s).
1. A method comprising:
at a query coordinator:
receiving, by one or more processors, a request from a computing resource for one or more files to process;
grouping a set of files from a shared file queue into a batch;
assigning the batch a unique batch ID;
storing, in a first staging area, the unique batch ID and information related to the batch;
assigning the batch to the computing resource;
receiving a file registration request from the computing resource, the file registration request including the unique batch ID and indicating that the batch has been processed; and
at the computing resource:
upon receiving the assignment of the batch, storing the unique batch ID and information related to the batch in a second staging area;
performing a first operator on the batch;
based on performing the first operator, generating a checkpoint signal;
determining a state condition of a second operator;
based on the state condition of the second operator, performing the second operator and passing the checkpoint signal to a third operator;
determining a state condition of the third operator;
based on the state condition of the third operator, clearing data associated with the third operator and performing the third operator;
finalizing the stored information in the second staging area;
generating a materialized result file as an output file associated with the batch;
uploading the materialized result file to a storage area from where the materialized result file is accessible to be scanned by the query coordinator; and
uploading a listing of the materialized result to an output shared file queue.
2. The method of claim 1 , further comprising:
deleting the unique batch ID and information related to the batch stored in the first staging area;
at the computing resource:
transmitting the output file associated with the batch;
generating the file registration request; and
deleting the unique batch ID and information related to the batch stored in the second staging area.
3. The method of claim 1 , wherein the first operator includes a table scan operator.
4. The method of claim 1 , wherein the third operator includes an insert operator.
5. The method of claim 2 , wherein the output file includes data corresponding only to the batch and no other batch.
6. The method of claim 1 , further comprising:
storing, in the first staging area, information related to a second batch including a unique batch ID for the second batch; and
upon not receiving a second file registration request associated with the second batch, rescheduling processing of the second batch based on the information related to the second batch stored in the first staging area.
7. The method of claim 6 , wherein rescheduling processing of the second batch includes assigning the second batch a different unique batch ID.
8. The method of claim 1 , wherein the file registration request includes information about an output file for the batch.
9. A system comprising:
one or more processors of a machine; and
a memory storing instructions that, when executed by the one or more processors, cause the machine to perform operations comprising:
at a query coordinator:
receiving a request from a computing resource for one or more files to process;
grouping a set of files from a shared file queue into a batch;
assigning the batch a unique batch ID;
storing, in a first staging area, the unique batch ID and information related to the batch;
assigning the batch to the computing resource;
receiving a file registration request from the computing resource, the file registration request including the unique batch ID and indicating that the batch has been processed; and
at the computing resource:
upon receiving the assignment of the batch, storing the unique batch ID and information related to the batch in a second staging area;
performing a first operator on the batch;
based on performing the first operator, generating a checkpoint signal;
determining a state condition of a second operator;
based on the state condition of the second operator, performing the second operator and passing the checkpoint signal to a third operator;
determining a state condition of the third operator;
based on the state condition of the third operator, clearing data associated with the third operator and performing the third operator;
finalizing the stored information in the second staging area;
generating a materialized result file as an output file associated with the batch;
uploading the materialized result file to a storage area from where the materialized result file is accessible to be scanned by the query coordinator; and
uploading a listing of the materialized result to an output shared file queue.
10. The system of claim 9 , the operations further comprising:
deleting the unique batch ID and information related to the batch stored in the first staging area;
at the computing resource:
transmitting the output file associated with the batch;
generating the file registration request; and
deleting the unique batch ID and information related to the batch stored in the second staging area.
11. The system of claim 9 , wherein the first operator includes a table scan operator.
12. The system of claim 9 , wherein the third operator includes an insert operator.
13. The system of claim 10 , wherein the output file includes data corresponding only to the batch and no other batch.
14. The system of claim 9 , the operations further comprising:
storing, in the first staging area, information related to a second batch including a unique batch ID for the second batch; and
upon not receiving a second file registration request associated with the second batch, rescheduling processing of the second batch based on the information related to the second batch stored in the first staging area.
15. The system of claim 14 , wherein rescheduling processing of the second batch includes assigning the second batch a different unique batch ID.
16. The system of claim 9 , wherein the file registration request includes information about an output file for the batch.
17. A machine-storage medium embodying instructions that, when executed by a machine, cause the machine to perform operations comprising:
at a query coordinator:
receiving a request from a computing resource for one or more files to process;
grouping a set of files from a shared file queue into a batch;
assigning the batch a unique batch ID;
storing, in a first staging area, the unique batch ID and information related to the batch;
assigning the batch to the computing resource;
receiving a file registration request from the computing resource, the file registration request including the unique batch ID and indicating that the batch has been processed; and
at the computing resource:
upon receiving the assignment of the batch, storing the unique batch ID and information related to the batch in a second staging area;
performing a first operator on the batch;
based on performing the first operator, generating a checkpoint signal;
determining a state condition of a second operator;
based on the state condition of the second operator, performing the second operator and passing the checkpoint signal to a third operator;
determining a state condition of the third operator;
based on the state condition of the third operator, clearing data associated with the third operator and performing the third operator;
finalizing the stored information in the second staging area;
generating a materialized result file as an output file associated with the batch;
uploading the materialized result file to a storage area from where the materialized result file is accessible to be scanned by the query coordinator; and
uploading a listing of the materialized result to an output shared file queue.
18. The machine-storage medium of claim 17 , further comprising:
deleting the unique batch ID and information related to the batch stored in the first staging area;
at the computing resource:
transmitting the output file associated with the batch;
generating the file registration request; and
deleting the unique batch ID and information related to the batch stored in the second staging area.
19. The machine-storage medium of claim 17 , wherein the first operator includes a table scan operator.
20. The machine-storage medium of claim 17 , wherein the third operator includes an insert operator.
21. The machine-storage medium of claim 18 , wherein the output file includes data corresponding only to the batch and no other batch.
22. The machine-storage medium of claim 17 , further comprising:
storing, in the first staging area, information related to a second batch including a unique batch ID for the second batch; and
upon not receiving a second file registration request associated with the second batch, rescheduling processing of the second batch based on the information related to the second batch stored in the first staging area.
23. The machine-storage medium of claim 22 , wherein rescheduling processing of the second batch includes assigning the second batch a different unique batch ID.
24. The machine-storage medium of claim 17 , wherein the file registration request includes information about an output file for the batch.