Systems, methods, and devices for managing data skew in a join operation
Systems, methods, and devices, for managing data skew during a join operation are disclosed. A method includes computing a hash value for a join operation and detecting data skew on a probe side of the join operation at a runtime of the join operation using a lightweight sketch data structure. The method includes identifying a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation. The method includes identifying a frequent build-side row having a build-side join key corresponding with the frequent probe-side join key. The method includes asynchronously distributing the frequent build-side row to one or more remote servers.
1. A method comprising:
detecting, by a local server comprising at least one hardware processor, data skew on a probe side of a join operation at a runtime of the join operation using a data structure;
identifying, by the local server, a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation;
identifying, by the local server, a frequent build-side row having a frequent build-side join key corresponding to the identified frequent probe-side join key; and
asynchronously distributing, by the local server in response to detecting the data skew on the probe side of the join operation, the identified frequent build-side row to a plurality of remote servers, the asynchronously distributing the frequent build-side row to the plurality of remote servers comprising broadcasting the frequent build-side row only to plurality of remote servers that frequently transmitted the frequent build-side join key.
2. The method of claim 1 , wherein each of the plurality of remote servers is configured to generate a separate hash table for the identified frequent build-side row.
3. The method of claim 1 , further comprising:
computing, by the local server, a hash value for the join operation;
selecting, by the local server, a rowset comprising a plurality of rows of the join operation; and
probing, by the local server, each of the plurality of rows of the rowset into a space saving algorithm using the hash value for the join operation.
4. The method of claim 3 , further comprising:
updating, by the local server, the space saving algorithm based on incoming data; and
identifying, by the local server for each update to the space saving algorithm, a frequency indicating how frequently the identified frequent probe-side join key is probed as a side-effect of the updating of the space saving algorithm.
5. The method of claim 4 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers is also in response to the identified frequency exceeding a predetermined threshold.
6. The method of claim 3 , further comprising:
calculating, by the local server, a total number of rows of the join operation that have been probed into the space saving algorithm;
calculating, by the local server, a threshold per worker thread based on the total number of rows of the join operation that have been probed into the space saving algorithm; and
determining, by the local server based on the threshold per worker thread, whether the frequent build-side join key is frequent among all threads of at least one server among the local server and the plurality of remote servers.
7. The method of claim 3 , wherein the data structure comprises a hash table space saving algorithm.
8. The method of claim 1 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers occurs only after determining, by the local server to a threshold confidence level, that the identified frequent probe-side join key is frequent on the local server.
9. The method of claim 1 , wherein asynchronously distributing the frequent build-side row to the plurality of remote servers comprises broadcasting the frequent build-side row to each of a plurality of available remote servers.
10. A local server comprising:
at least one hardware processor; and
one or more non-transitory computer-readable storage media containing instructions that, when executed by the at least one hardware processor, cause the at least one hardware processor to perform operations comprising:
detecting data skew on a probe side of a join operation at a runtime of the join operation using a data structure;
identifying a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation;
identifying a frequent build-side row having a frequent build-side join key corresponding to the identified frequent probe-side join key; and
asynchronously distributing, in response to detecting the data skew on the probe side of the join operation, the identified frequent build-side row to a plurality of remote servers, the asynchronously distributing the frequent build-side row to the plurality of remote servers comprising broadcasting the frequent build-side row only to plurality of remote servers that frequently transmitted the frequent build-side join key.
11. The local server of claim 10 , wherein each of the plurality of remote servers is configured to generate a separate hash table for the identified frequent build-side row.
12. The local server of claim 10 , the operations further comprising:
computing, by the local server, a hash value for the join operation;
selecting, by the local server, a rowset comprising a plurality of rows of the join operation; and
probing, by the local server, each of the plurality of rows of the rowset into a space saving algorithm using the hash value for the join operation.
13. The local server of claim 12 , the operations further comprising:
updating, by the local server, the space saving algorithm based on incoming data; and
identifying, by the local server for each update to the space saving algorithm, a frequency indicating how frequently the identified frequent probe-side join key is probed as a side-effect of the updating of the space saving algorithm.
14. The local server of claim 13 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers is also in response to the identified frequency exceeding a predetermined threshold.
15. The local server of claim 12 , the operations further comprising:
calculating, by the local server, a total number of rows of the join operation that have been probed into the space saving algorithm;
calculating, by the local server, a threshold per worker thread based on the total number of rows of the join operation that have been probed into the space saving algorithm; and
determining, by the local server based on the threshold per worker thread, whether the frequent build-side join key is frequent among all threads of at least one server among the local server and the plurality of remote servers.
16. The local server of claim 12 , wherein the data structure comprises a hash table space saving algorithm.
17. The local server of claim 10 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers occurs only after determining, by the local server to a threshold confidence level, that the identified frequent probe-side join key is frequent on the local server.
18. The local server of claim 10 , wherein asynchronously distributing the frequent build-side row to the plurality of remote servers comprises broadcasting the frequent build-side row to each of a plurality of available remote servers.
19. One or more non-transitory computer-readable storage media containing instructions that, when executed by at least one hardware processor, cause the at least one hardware processor to perform operations comprising:
detecting, by a local server, data skew on a probe side of a join operation at a runtime of the join operation using a data structure;
identifying, by the local server, a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation;
identifying, by the local server, a frequent build-side row having a frequent build-side join key corresponding to the identified frequent probe-side join key; and
asynchronously distributing, by the local server in response to detecting the data skew on the probe side of the join operation, the identified frequent build-side row to a plurality of remote servers, the asynchronously distributing the frequent build-side row to the plurality of remote servers comprising broadcasting the frequent build-side row only to plurality of remote servers that frequently transmitted the frequent build-side join key.
20. The one or more non-transitory computer-readable storage media of claim 19 , wherein each of the plurality of remote servers is configured to generate a separate hash table for the identified frequent build-side row.
21. The one or more non-transitory computer-readable storage media of claim 19 , the operations further comprising:
computing, by the local server, a hash value for the join operation;
selecting, by the local server, a rowset comprising a plurality of rows of the join operation; and
probing, by the local server, each of the plurality of rows of the rowset into a space saving algorithm using the hash value for the join operation.