System and method for precise quantile determination

1. A method for precise quantile determination in a system in which data is distributed among nodes in the system, one of a plurality of worker nodes functions as a coordinator node, each of the worker nodes having local data for the precise quantile determination, the method comprising:

by each processor in parallel at each of the worker nodes, determining a number N of datapoints to be computed from the local data of the worker node;

by one of the processors at one of the nodes, selecting as the coordinator node one of the nodes with a largest number of datapoints;

by a processor at the coordinator node, determining from the number N of datapoints and one or more quantiles to be computed, a maximum relative error value c so that a rank distance between two adjacent elements in all of the one or more quantiles is at most 2ε, wherein the maximum relative error value c guarantees that an exact result for every quantile is determined in two passes, wherein the value for the maximum relative error ε is determined by solving the following equations:

M approx =( b+ 3)3 dL+hL

M exact =4ε N d M

M approx =M exact ,

such that M approx and M exact are memory consumptions, M is the number of requested quantiles, b is a block size for partitioning input buffers, d is memory per entry, L is a number of levels in the quantile summary, h is additional overhead for pointers that connect the levels in the quantile summary, and N is an amount of the datapoints;

by each processor in parallel at each of the worker nodes, computing, in a first pass of the datapoints of the local data of each of the worker nodes, using an approximation algorithm and the maximum relative error value ε, a quantile summary of the datapoints of the local data that specifies local lower and upper bounds of the quantile ranks for selected datapoints of the local data of each of the worker nodes;

by the processor at the coordinator node:

merging the quantile summaries for the datapoints of the local data of each of the worker nodes, into a constructed summary;

determining, based on the constructed summary, for each quantile rank, second-pass lower and upper bounds for values between which the quantile must occur;

by each processor in parallel at each of the worker nodes, filtering out, in a second pass over the datapoints of the local data at each of the worker nodes, the datapoints that are not between the second-pass lower and upper bounds determined at the coordinator node based on the constructed summary which is based on all of the plurality of worker nodes, to provide subsets of datapoints of the local data at the worker node which are within the second-pass lower and upper bounds from the coordinator node;

by the processor at the coordinator node:

collecting, from the worker nodes, as collected qualifying data, all of the subsets of datapoints of the local data from the worker nodes which is within the second-pass lower and upper bounds determined at the coordinator node; and

computing, based on the quantile ranks and the collected qualifying data which were collected from the worker nodes, the precise quantiles.

2. The method of claim 1 , further comprising streaming, by the worker nodes, the subsets of datapoints of the local data within the lower and upper bounds, to the coordinator node.

3. The method of claim 1 , wherein when the lower and upper bounds for all quantiles are identical, then the precise quantile determination is complete.

4. The method of claim 1 , wherein

the filtering out step further comprises, by each processor in parallel at each of the worker nodes:

collecting the datapoints that lie between the lower and upper bounds;

counting/excluding the datapoints that are on or below the lower bound; and

the collecting and the computing by the coordinator node further comprises:

sorting the datapoints that lie between the lower and upper bounds;

selecting a datapoint of a requested rank from the sorted datapoints; and

simultaneously correcting the requested rank by the number of datapoints on or below the lower bound.

5. A system for precise quantile determination in which data is distributed among nodes in the system, one of a plurality of worker nodes functions as a coordinator node, each of the worker nodes having local data for the precise quantile determination comprising:

a coordinator node having

a memory; and

a processor cooperatively operable with the memory; and

a plurality of worker nodes, each worker node having

a memory; and

a processor cooperatively operable with the memory,

the processor of the coordinator node is configured to, based on instructions stored in the memory of the coordinator node, determine from the number N of datapoints reported from the worker nodes and one or more quantiles to be computed, a maximum relative error value c so that a rank distance between two adjacent elements in all of the one or more quantiles is at most 2ε, wherein the maximum relative error value ε guarantees that an exact result for every quantile is determined in two passes, wherein the value for the maximum relative error ε is determined by solving the following equations:

M approx =( b+ 3)3 dL+hL

M exact =4ε N d M

M approx =M exact ,

such that M approx and M exact are memory consumptions, M is the number of requested quantiles, b is a block size for partitioning input buffers, d is memory per entry, L is a number of levels in the quantile summary, h is additional overhead for pointers that connect the levels in the quantile summary, and N is an amount of datapoints;

each processor of each of the worker nodes is configured to operate in parallel with the other worker nodes, and, based on instructions stored in the memory of the worker node, compute, in a first pass of the datapoints of the local data of each of the worker nodes, using an approximation algorithm and the maximum relative error value ε, a quantile summary of the datapoints of the local data that specifies local lower and upper bounds of the quantile ranks for selected datapoints of the local data of each of the worker nodes,

the processor at the coordinator node is further configured to

merge the quantile summaries for the datapoints of the local data of each of the worker nodes, into a constructed summary;

determine, based on the constructed summary, for each quantile rank, second-pass lower and upper bounds for values between which the quantile must occur;

each processor of each of the worker nodes is further configured to filter out, in a second pass over the datapoints of the local data at each of the worker nodes, the datapoints that are not between the second-pass lower and upper bounds determined at the coordinator node based on the constructed summary which is based on all of the plurality of worker nodes, to provide subsets of datapoints of the local data at the worker node which are within the second-pass lower and upper bounds from the coordinator node;

the processor at the coordinator node is further configured to

collect, from the worker nodes, as collected qualifying data, all of the subsets of datapoints of the local data from the worker nodes which is within the second-pass lower and upper bounds determined at the coordinator node; and

compute, based on the quantile ranks and the collected qualifying data which were collected from the worker nodes, the precise quantiles.

6. The system of claim 5 , wherein the processor of each of the worker nodes is further configured to stream the subsets of datapoints of the local data within the lower and upper bounds, to the coordinator node.

7. The system of claim 5 , wherein when the lower and upper bounds for all quantiles are identical, then the precise quantile determination is complete.

8. The system of claim 5 , wherein the processor at the worker nodes filters out from the datapoints those datapoints that are not between the lower and upper bounds by:

collecting the datapoints that lie between the lower and upper bounds;

counting/excluding the datapoints that are on or below the lower bound;

sorting the datapoints that lie between the lower and upper bounds;

selecting a datapoint of a requested rank from the sorted datapoints; and

simultaneously correcting the requested rank by the number of datapoints on or below the lower bound.

9. A non-transitory computer readable medium comprising executable instructions for a method for precise quantile determination in a system in which data is distributed among nodes in the system, one of a plurality of worker nodes functions as a coordinator node, each of the worker nodes having local data for the precise quantile determination, the instructions being executed on a processor to:

determine, by each processor in parallel at each of the worker nodes, a number N of datapoints to be computed from local data of each worker node;

select, by one of the processors at one of the nodes, as the coordinator node, one of the nodes with a largest number of datapoints;

determine, by a processor at the coordinator node, from the number N of datapoints and one or more quantiles to be computed, a maximum relative error value ε so that a rank distance between two adjacent elements in all of the one or more quantiles is at most 2ε, wherein the maximum relative error value ε guarantees that an exact result for every quantile is determined in two passes, wherein the instructions are further executed so that the value for the maximum relative error ε is determined by solving the following equations:

M approx =( b+ 3)3 dL+hL

M exact =4ε N d M

M approx =M exact ,

such that M approx and M exact are memory consumptions, M is the number of requested quantiles, b is a block size for partitioning input buffers, d is memory per entry, L is a number of levels in the quantile summary, h is additional overhead for pointers that connect the levels in the quantile summary, and N is an amount of the datapoints;

compute, by each processor in parallel at each of the worker nodes, in a first pass of the datapoints of the local data of each the worker nodes, using an approximation algorithm and the maximum relative error ε, a quantile summary of the datapoints that specifies local lower and upper bounds of the quantile ranks for selected datapoints of the local data of each of the worker nodes;

merge, by the processor at the coordinator node, the quantile summaries for the datapoints of the local data of each of the worker nodes, into a constructed summary;

determine, by the processor at the coordinator node, based on the constructed summary, for each quantile rank, second-pass lower and upper bounds for values between which the quantile must occur;

filter out, by each processor in parallel at the worker nodes, in a second pass over the datapoints of the local data at each of the worker nodes, the datapoints that are not between the second-pass lower and upper bounds determined at the coordinator node based on the constructed summary which is based on all of the plurality of worker nodes, to provide subsets of datapoints of the local data at the worker node which are within the second-pass lower and upper bounds from the coordinator node; and

collect, by the processor at the coordinator node, from the worker nodes, as collected qualifying data, all of the subsets of datapoints of the local data from the worker nodes which is within the second-pass lower and upper bounds determined at the coordinator node; and

compute, by the processor at the coordinator node, based on the quantile ranks and the collected qualifying data which were collected from the worker nodes, the precise quantiles.

10. The non-transitory computer readable medium of claim 9 , wherein the instructions are further executed to stream, by the worker nodes, the subsets of datapoints of the local data within the lower and upper bounds, to the coordinator node.

11. The non-transitory computer readable medium of claim 9 , wherein the instructions are further executed so that when the lower and upper bounds for all quantiles are identical, then the precise quantile determination is complete.

12. The non-transitory computer readable medium of claim 9 , wherein the filtering out instructions are further executed to, by the processor at each of the worker nodes:

collect the datapoints that lie between the lower and upper bounds;

count/exclude the datapoints that are on or below the lower bound;

sort the datapoints that lie between the lower and upper bounds;

select a datapoint of a requested rank from the sorted datapoints; and

simultaneously correct the requested rank by the number of datapoints on or below the lower bound.