Distributed columnar data set and metadata storage
An apparatus includes a processor to: within each collection thread, assemble a row group from stored rows, generate row group metadata corresponding to the row group, and store the row group and row group metadata within a data buffer of a queue; operate the queue as a FIFO buffer; within each aggregation thread, retrieve multiple row groups and corresponding row group metadata from multiple data buffers of the queue, assemble a data set part from the multiple row groups, generate part metadata that includes the row group metadata, and transmit, to storage device(s) and/or a requesting device, the data set part and/or the part metadata; and in response to each retrieval of at least a row group from a data buffer of the queue for an aggregation thread, analyze availability of storage space within the node device to determine whether to dynamically adjust the quantity of data buffers.
1. An apparatus comprising at least one processor and a storage to store instructions that, when executed by the at least one processor, cause the at least one processor to perform operations comprising:
receive, at a node device of multiple node devices, and from a control device via a network, an instruction to the multiple node devices to persistently store a data set within at least one storage device, wherein:
the data set comprises multiple data values organized into numerous rows;
each row comprises multiple data fields that each fall within a column of multiple columns;
the numerous rows of the data set are divided into subsets of multiple rows that are distributed among multiple storage spaces provided by the multiple node devices; and
the data values of the multiple rows that are stored within the storage space provided by each node device are stored in a row-wise organization in which data values within each row are stored at adjacent storage locations;
in response to receiving the instruction to persistently store the multiple rows that are stored within the storage space provided by the node device, the at least one processor is caused to perform operations comprising:
within each collection thread of a quantity of collection threads, the at least one processor is caused to perform operations comprising:
assemble a subset of the multiple rows stored within the node device into a row group with the data values reorganized into a columnar organization;
generate row group metadata corresponding to the row group that includes, for each column, indications of the highest and lowest data values, and each unique data value; and
store the row group and row group metadata within a data buffer of a buffer queue;
operate the buffer queue as a first-in-first-out (FIFO) buffer in which the first data buffer of multiple data buffers to be filled with a row group from a collection thread becomes the first data buffer from which a row group is retrieved by an aggregation thread;
within each aggregation thread of a quantity of aggregation threads, the at least one processor is caused to perform operations comprising:
assemble a data set part from multiple row groups retrieved from multiple data buffers of the buffer queue;
generate part metadata corresponding to the data set part from multiple row group metadata corresponding to the multiple row groups, and retrieved from the multiple data buffers; and
transmit at least one of the data set part and the part metadata to the at least one storage device via the network; and
in response to each instance of retrieval of a row group and corresponding row group metadata from a data buffer of the buffer queue for use within an aggregation thread, analyze a level of availability of storage space within the node device to determine whether to dynamically adjust a quantity of data buffers of the buffer queue.
2. The apparatus of claim 1 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
for each column within the row group, generate a hash value from each identified unique data value, and identify highest and lowest hash values; and
include, in the corresponding row group metadata, indications of the highest and lowest hash values for each column.
3. The apparatus of claim 1 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
for each column within the row group, identify highest and lowest data values, and each unique data value; and
include, in the corresponding row group metadata, indications of the highest and lowest data values, and each unique data value for each column.
4. The apparatus of claim 1 , wherein, within each aggregation thread of the quantity of aggregation threads, the at least one processor is caused to perform operations comprising:
retrieve the multiple row groups and the multiple row group metadata from the multiple data buffers;
from the indications of the row group metadata corresponding to all of the row groups of the multiple row groups, identify the highest and lowest data values in each column across all of the row groups of the multiple row groups; and
include, in the part metadata, indications of the highest and lowest data values in each column across all of the row groups of the multiple row groups.
5. The apparatus of claim 1 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
reorganize the data values within the row group from the row-wise organization to the columnar organization, wherein the data values within each column of the multiple columns within the row group are stored at adjacent storage locations within the storage space provided by the node device;
losslessly compress the data values within at least one column of the multiple columns within the row group; and
include, in the corresponding row group metadata, an indication of at least one of a type of lossless compression used to losslessly compress the data values within the at least one column, and an identifier of the at least one column.
6. The apparatus of claim 1 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
reorganize the data values within the row group from the row-wise organization to the columnar organization, wherein the data values within each column of the multiple columns within the row group are stored at adjacent storage locations within the storage space provided by the node device;
encrypt the data values within at least one column of the multiple columns within the row group; and
include, in the corresponding row group metadata, an indication of at least one of a type of encryption used to encrypt the data values within the at least one column, and an identifier of the at least one column.
7. The apparatus of claim 1 , wherein the at least one processor is caused to perform operations comprising:
instantiate the quantity of collection threads of execution, the quantity of data buffers of a buffer queue, and the quantity of aggregation threads of execution; and
in response to each instance of storage of a row group into a data buffer of the buffer queue from a collection thread, analyze at least one of the level of availability of storage space within the node device and a level of availability of processing resources of the at least one processor to determine whether to dynamically adjust the quantity of collection threads.
8. The apparatus of claim 1 , wherein the at least one processor is caused to perform operations comprising:
instantiate the quantity of collection threads of execution, the quantity of data buffers of a buffer queue, and the quantity of aggregation threads of execution; and
in response to each instance of transmission of a data set part to the at least one storage device from an aggregation thread, analyze at least one of the level of availability of storage space within the node device and a level of availability of processing resources of the at least one processor to determine whether to dynamically adjust the quantity of aggregation threads.
9. The apparatus of claim 1 , wherein:
each data set part of the multiple data set parts is to be persistently stored by the at least one storage device within a separate data file of multiple data files;
the at least one storage device is to maintain a subdirectory in which the multiple data files are to be stored;
the control device is to persistently store a metadata object within the subdirectory that includes the part metadata corresponding to all of the multiple data set parts of the data set;
an identifier that uniquely identifies the data set is to be used by the at least one storage device to identify the subdirectory; and
within each aggregation thread, the at least one processor is caused to perform operations comprising:
transmit, to the control device via the network, a request for a sequence number;
receive, from the control device via the network, the sequence number;
combine the sequence number received from the control device with the identifier of the data set, and with an identifier of a file type of the data file in which the data set part is to be persistently stored, to form a file identifier to be used by the at least one storage device to uniquely identify the data file in which the data set part is to be persistently stored; and
transmit, to the at least one storage device via the network, the data set part along with the file identifier.
10. The apparatus of claim 9 , wherein:
the control device includes a columnar schema in the metadata object prior to persistently storing the metadata object; and
the columnar schema specifies at least one of:
a quantity of rows to be included in each row group;
a quantity of row groups to be included in each data set part;
a data type that data values within a column of each row group are to be normalized to;
a data format that data values within a column of each row group are to be normalized to;
a type of compression that is to be performed on a column within each row group;
a type of compression that is to be performed on each whole row group;
a type of compression that is to be performed on each data set part;
a type of encryption that is to be performed on a column within each row group;
a type of encryption that is to be performed on each whole row group; and
a type of encryption that is to be performed on each data set part.
11. A computer-program product tangibly embodied in a non-transitory machine-readable storage medium, the computer-program product including instructions operable to cause at least one processor of a node device of multiple node devices to perform operations comprising:
receive, at the node device of multiple node devices, and from a control device via a network, an instruction to the multiple node devices to persistently store a data set within at least one storage device, wherein:
the data set comprises multiple data values organized into numerous rows;
each row comprises multiple data fields that each fall within a column of multiple columns;
the numerous rows of the data set are divided into subsets of multiple rows that are distributed among multiple storage spaces provided by the multiple node devices; and
the data values of the multiple rows that are stored within the storage space provided by each node device are stored in a row-wise organization in which data values within each row are stored at adjacent storage locations;
in response to receiving the instruction to persistently store the multiple rows that are stored within the storage space provided by the node device, the at least one processor is caused to perform operations comprising:
within each collection thread of a quantity of collection threads, the at least one processor is caused to perform operations comprising:
assemble a subset of the multiple rows stored within the node device into a row group with the data values reorganized into a columnar organization;
generate row group metadata corresponding to the row group that includes, for each column, indications of the highest and lowest data values, and each unique data value; and
store the row group and row group metadata within a data buffer of a buffer queue;
operate the buffer queue as a first-in-first-out (FIFO) buffer in which the first data buffer of multiple data buffers to be filled with a row group from a collection thread becomes the first data buffer from which a row group is retrieved by an aggregation thread;
within each aggregation thread of a quantity of aggregation threads, the at least one processor is caused to perform operations comprising:
assemble a data set part from multiple row groups retrieved from multiple data buffers of the buffer queue;
generate part metadata corresponding to the data set part from multiple row group metadata corresponding to the multiple row groups, and retrieved from the multiple data buffers; and
transmit at least one of the data set part and the part metadata to the at least one storage device via the network; and
in response to each instance of retrieval of a row group and corresponding row group metadata from a data buffer of the buffer queue for use within an aggregation thread, analyze a level of availability of storage space within the node device to determine whether to dynamically adjust a quantity of data buffers of the buffer queue.
12. The computer-program product of claim 11 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
for each column within the row group, generate a hash value from each identified unique data value, and identify highest and lowest hash values; and
include, in the corresponding row group metadata, indications of the highest and lowest hash values for each column.
13. The computer-program product of claim 11 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
for each column within the row group, identify highest and lowest data values, and each unique data value; and
include, in the corresponding row group metadata, indications of the highest and lowest data values, and each unique data value for each column.
14. The computer-program product of claim 11 , wherein, within each aggregation thread of the quantity of aggregation threads, the at least one processor is caused to perform operations comprising:
retrieve the multiple row groups and the multiple row group metadata from the multiple data buffers;
from the indications of the row group metadata corresponding to all of the row groups of the multiple row groups, identify the highest and lowest data values in each column across all of the row groups of the multiple row groups; and
include, in the part metadata, indications of the highest and lowest data values in each column across all of the row groups of the multiple row groups.
15. The computer-program product of claim 11 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
reorganize the data values within the row group from the row-wise organization to the columnar organization, wherein the data values within each column of the multiple columns within the row group are stored at adjacent storage locations within the storage space provided by the node device;
losslessly compress the data values within at least one column of the multiple columns within the row group; and
include, in the corresponding row group metadata, an indication of at least one of a type of lossless compression used to losslessly compress the data values within the at least one column, and an identifier of the at least one column.
16. The computer-program product of claim 11 , wherein, within each collection thread of the quantity of collection threads, the at least one processor is caused to perform operations comprising:
reorganize the data values within the row group from the row-wise organization to the columnar organization, wherein the data values within each column of the multiple columns within the row group are stored at adjacent storage locations within the storage space provided by the node device;
encrypt the data values within at least one column of the multiple columns within the row group; and
include, in the corresponding row group metadata, an indication of at least one of a type of encryption used to encrypt the data values within the at least one column, and an identifier of the at least one column.
17. The computer-program product of claim 11 , wherein the at least one processor is caused to perform operations comprising:
instantiate the quantity of collection threads of execution, the quantity of data buffers of a buffer queue, and the quantity of aggregation threads of execution; and
in response to each instance of storage of a row group into a data buffer of the buffer queue from a collection thread, analyze at least one of the level of availability of storage space within the node device and a level of availability of processing resources of the at least one processor to determine whether to dynamically adjust the quantity of collection threads.
18. The computer-program product of claim 11 , wherein the at least one processor is caused to perform operations comprising:
instantiate the quantity of collection threads of execution, the quantity of data buffers of a buffer queue, and the quantity of aggregation threads of execution; and
in response to each instance of transmission of a data set part to the at least one storage device from an aggregation thread, analyze at least one of the level of availability of storage space within the node device and a level of availability of processing resources of the at least one processor to determine whether to dynamically adjust the quantity of aggregation threads.
19. The computer-program product of claim 11 , wherein:
each data set part of the multiple data set parts is to be persistently stored by the at least one storage device within a separate data file of multiple data files;
the at least one storage device is to maintain a subdirectory in which the multiple data files are to be stored;
the control device is to persistently store a metadata object within the subdirectory that includes the part metadata corresponding to all of the multiple data set parts of the data set;
an identifier that uniquely identifies the data set is to be used by the at least one storage device to identify the subdirectory; and
within each aggregation thread, the at least one processor is caused to perform operations comprising:
transmit, to the control device via the network, a request for a sequence number;
receive, from the control device via the network, the sequence number;
combine the sequence number received from the control device with the identifier of the data set, and with an identifier of a file type of the data file in which the data set part is to be persistently stored, to form a file identifier to be used by the at least one storage device to uniquely identify the data file in which the data set part is to be persistently stored; and
transmit, to the at least one storage device via the network, the data set part along with the file identifier.
20. The computer-program product of claim 19 , wherein:
the control device includes a columnar schema to the metadata object prior to persistently storing the metadata object; and
the columnar schema specifies at least one of:
a quantity of rows to be included in each row group;
a quantity of row groups to be included in each data set part;
a data type that data values within a column of each row group are to be normalized to;
a data format that data values within a column of each row group are to be normalized to;
a type of compression that is to be performed on a column within each row group;
a type of compression that is to be performed on each whole row group;
a type of compression that is to be performed on each data set part;
a type of encryption that is to be performed on a column within each row group;
a type of encryption that is to be performed on each whole row group; and
a type of encryption that is to be performed on each data set part.
21. A computer-implemented method comprising:
receiving, at a node device of multiple node devices, and from a control device via a network, an instruction to the multiple node devices to persistently store a data set within at least one storage device, wherein:
the data set comprises multiple data values organized into numerous rows;
each row comprises multiple data fields that each fall within a column of multiple columns;
the numerous rows of the data set are divided into subsets of multiple rows that are distributed among multiple storage spaces provided by the multiple node devices; and
the data values of the multiple rows that are stored within the storage space provided by each node device are stored in a row-wise organization in which data values within each row are stored at adjacent storage locations;
in response to receiving the instruction to persistently store the multiple rows that are stored within the storage space provided by the node device, performing operations comprising:
within each collection thread of a quantity of collection threads, at least one processor of the node device is caused to perform operations comprising:
assembling, by the at least one processor, a subset of the multiple rows stored within the node device into a row group with the data values reorganized into a columnar organization;
generating, by the at least one processor, row group metadata corresponding to the row group that includes, for each column, indications of the highest and lowest data values, and each unique data value; and
storing the row group and row group metadata within a data buffer of a buffer queue;
operating, by the at least one processor, the buffer queue as a first-in-first-out (FIFO) buffer in which the first data buffer of multiple data buffers to be filled with a row group from a collection thread becomes the first data buffer from which a row group is retrieved by an aggregation thread;
within each aggregation thread of a quantity of aggregation threads, the at least one processor is caused to perform operations comprising:
assembling, by the at least one processor, a data set part from multiple row groups retrieved from multiple data buffers of the buffer queue;
generating, by the at least one processor, part metadata corresponding to the data set part from multiple row group metadata corresponding to the multiple row groups, and retrieved from the multiple data buffers; and
transmitting, from the at least one processor, at least one of the data set part and the part metadata to the at least one storage device via the network; and
in response to each instance of retrieval of a row group and corresponding row group metadata from a data buffer of the buffer queue for use within an aggregation thread, analyzing, by the at least one processor, a level of availability of storage space within the node device to determine whether to dynamically adjust a quantity of data buffers of the buffer queue.
22. The method of claim 21 , comprising, within each collection thread of the quantity of collection threads, performing operations comprising:
for each column within the row group, generating a hash value from each identified unique data value, and identifying highest and lowest hash values; and
including, in the corresponding row group metadata, indications of the highest and lowest hash values for each column.
23. The method of claim 21 , comprising, within each collection thread of the quantity of collection threads, performing operations comprising:
for each column within the row group, identifying highest and lowest data values, and each unique data value; and
including, in the corresponding row group metadata, indications of the highest and lowest data values, and each unique data value for each column.
24. The method of claim 21 , comprising, within each aggregation thread of the quantity of aggregation threads, performing operations comprising:
retrieving the multiple row groups and the multiple row group metadata from the multiple data buffers;
from the indications of the row group metadata corresponding to all of the row groups of the multiple row groups, identifying the highest and lowest data values in each column across all of the row groups of the multiple row groups; and
including, in the part metadata, indications of the highest and lowest data values in each column across all of the row groups of the multiple row groups.
25. The method of claim 21 , comprising, within each collection thread of the quantity of collection threads, performing operations comprising:
reorganizing the data values within the row group from the row-wise organization to the columnar organization, wherein the data values within each column of the multiple columns within the row group are stored at adjacent storage locations within the storage space provided by the node device;
losslessly compressing the data values within at least one column of the multiple columns within the row group; and
including, in the corresponding row group metadata, an indication of at least one of a type of lossless compression used to losslessly compress the data values within the at least one column, and an identifier of the at least one column.
26. The method of claim 21 , comprising, within each collection thread of the quantity of collection threads, performing operations comprising:
reorganize the data values within the row group from the row-wise organization to the columnar organization, wherein the data values within each column of the multiple columns within the row group are stored at adjacent storage locations within the storage space provided by the node device;
encrypting the data values within at least one column of the multiple columns within the row group; and
including, in the corresponding row group metadata, an indication of at least one of a type of encryption used to encrypt the data values within the at least one column, and an identifier of the at least one column.
27. The method of claim 21 , comprising:
instantiating the quantity of collection threads of execution, the quantity of data buffers of a buffer queue, and the quantity of aggregation threads of execution; and
in response to each instance of storage of a row group into a data buffer of the buffer queue from a collection thread, analyzing at least one of the level of availability of storage space within the node device and a level of availability of processing resources of the at least one processor to determine whether to dynamically adjust the quantity of collection threads.
28. The method of claim 21 , comprising:
instantiating the quantity of collection threads of execution, the quantity of data buffers of a buffer queue, and the quantity of aggregation threads of execution; and
in response to each instance of transmission of a data set part to the at least one storage device from an aggregation thread, analyzing at least one of the level of availability of storage space within the node device and a level of availability of processing resources of the at least one processor to determine whether to dynamically adjust the quantity of aggregation threads.
29. The method of claim 21 , wherein:
each data set part of the multiple data set parts is to be persistently stored by the at least one storage device within a separate data file of multiple data files;
the at least one storage device is to maintain a subdirectory in which the multiple data files are to be stored;
the control device is to persistently store a metadata object within the subdirectory that includes the part metadata corresponding to all of the multiple data set parts of the data set;
an identifier that uniquely identifies the data set is to be used by the at least one storage device to identify the subdirectory; and
the method comprises, within each aggregation thread, performing operations comprising:
transmitting, to the control device via the network, a request for a sequence number;
receiving, from the control device via the network, the sequence number;
combining the sequence number received from the control device with the identifier of the data set, and with an identifier of a file type of the data file in which the data set part is to be persistently stored, to form a file identifier to be used by the at least one storage device to uniquely identify the data file in which the data set part is to be persistently stored; and
transmitting, to the at least one storage device via the network, the data set part along with the file identifier.
30. The method of claim 29 , wherein:
the control device includes a columnar schema to the metadata object prior to persistently storing the metadata object; and
the columnar schema specifies at least one of:
a quantity of rows to be included in each row group;
a quantity of row groups to be included in each data set part;
a data type that data values within a column of each row group are to be normalized to;
a data format that data values within a column of each row group are to be normalized to;
a type of compression that is to be performed on a column within each row group;
a type of compression that is to be performed on each whole row group;
a type of compression that is to be performed on each data set part;
a type of encryption that is to be performed on a column within each row group;
a type of encryption that is to be performed on each whole row group; and
a type of encryption that is to be performed on each data set part.