You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@hive.apache.org by "Navis (JIRA)" <ji...@apache.org> on 2012/07/20 09:51:40 UTC
[jira] [Created] (HIVE-3286) Explicit skew join on user provided
condition
Navis created HIVE-3286:
---------------------------
Summary: Explicit skew join on user provided condition
Key: HIVE-3286
URL: https://issues.apache.org/jira/browse/HIVE-3286
Project: Hive
Issue Type: Improvement
Components: Query Processor
Affects Versions: 0.10.0
Reporter: Navis
Assignee: Navis
Priority: Minor
Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
As for a start, I've extended join grammar something like this.
{code}
select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
{code}
which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13423735#comment-13423735 ]
Namit Jain commented on HIVE-3286:
----------------------------------
Otherwise, I think this is generic and is useful for Hive.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Nadeem Moidu (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13419378#comment-13419378 ]
Nadeem Moidu commented on HIVE-3286:
------------------------------------
Here is the other JIRA https://issues.apache.org/jira/browse/HIVE-3086 .
I'm not sure if using the list bucketing schema constricts you (the phrase "list bucketed by" has been removed from it). List bucketing was anyway solving a problem caused by skew, so there was no point expecting the user to give the skew information more than once.
This seems to be solving a slightly different problem, e.g. I don't allow ranges.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13423808#comment-13423808 ]
Navis commented on HIVE-3286:
-----------------------------
Most of our queries are driven by big table (500G~) joining small tables (~10G) and big table is heavily skewed with one or a few keys(more than 60%). In this case RANDOM distribution would be very useful in spite of additional cost of duplication. And for small tables, there is not that much for the keys, which minimizes overload of duplication. I'll post test results later if possible.
KEYS is distribution by join keys. EXPRESSIONs can be different from that, though should be composed of join keys. I also think this is not so useful option and even removed once. But I added it in final version for just in case.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Comment Edited] (HIVE-3286) Explicit skew join on user
provided condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13425679#comment-13425679 ]
Namit Jain edited comment on HIVE-3286 at 7/31/12 11:21 AM:
------------------------------------------------------------
Comments on phabricator
was (Author: namit):
Comments on pbabricator
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Navis updated HIVE-3286:
------------------------
Description:
Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
As for a start, I've extended join grammar something like this.
{code}
select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
{code}
which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
Work till done now will be updated shortly after code cleanup.
----------------------------
Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
"DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
2. KEYS : determined by hash value of keys (same with previous)
3. expression : determined by hash of object evaluated by user-provided expression
Only possible with inner, equi, common-joins. Not yet supports join tree merging.
Might be used by other RS users like "SORT BY" or "GROUP BY"
If there exists column statistics for the key, it could be possible to apply automatically.
For example, if 20 reducers are used for the query below,
{code}
select count(*) from src a join src b on a.key=b.key skew on (
a.key = '0' CLUSTER BY 10 PERCENT,
b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
{code}
group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
For a row with key='200', this is not belong to any skew group : hash(key) % 6
*expressions in skew condition :
1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
3. all functions in expression should be deteministic and stateless.
4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
**driver alias :
1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
was:
Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
As for a start, I've extended join grammar something like this.
{code}
select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
{code}
which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
Work till done now will be updated shortly after code cleanup.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13420100#comment-13420100 ]
Navis commented on HIVE-3286:
-----------------------------
This is for assigning some number of reducers exclusively for a key (or group of keys).
"SKEWED BY 30 PERCENT" means if the total number of reducer for MR is 20, hive assign 20*0.3=6 reducers for the group. If not specified, one reducer is assigned for that group. For above example, resultant partition number of group 1 is distributed in the range of 12~17, group 2 is 18, group 3 is 19, and remaining keys are distributed in the range of 0~11.
"DISTRIBUTED BY a.key-1" means if partition range is more than 1(like group 1), distribution in the range(12~17) is based on hash of evaluated value by the expression 1.key-1. I think this is not yet enough for real usage.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13423856#comment-13423856 ]
Navis commented on HIVE-3286:
-----------------------------
I cannot understand exactly what you said 'create group for each key'. If it's something described in https://cwiki.apache.org/Hive/skewed-join-optimization.html, I should say it's too difficult for developers who would translate sqls for rdbms to hsql(that's what I'm supporting).
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13425679#comment-13425679 ]
Namit Jain commented on HIVE-3286:
----------------------------------
Comments on pbabricator
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13424768#comment-13424768 ]
Namit Jain commented on HIVE-3286:
----------------------------------
Ya, why dont you remove it ?
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Navis updated HIVE-3286:
------------------------
Status: Patch Available (was: Open)
https://reviews.facebook.net/D4287
Slightly upgraded grammar, for example
{code}
select * from src a join src b on a.key=b.key skew on
(a.key+1 < 50 SKEWED BY 30 PERCENT DISTRIBUTE BY a.key-1,
a.key+1 < 100 SKEWED BY 20 PERCENT,
a.key < 150);
{code}
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13424728#comment-13424728 ]
Navis commented on HIVE-3286:
-----------------------------
Default is random, which should duplicate rows of small tables.
When skewing for each key is not so severe, distributing by join key can be enough without duplicating row. For example, logs of last two hour is multiple of others, user can make a group for them and distribute them again by join key.
{code}
select ~~ from logs join errors on logs.hour=errors.hour AND logs.error_seq = errors.error_seq
skew on (last_two_hour(logs.hour) cluster by 60 PERCENT DISTRIBUTE BY KEYS)
{code}
DISTRIBUTE BY <expression> can provides more control on key distribution. In above case, "DISTRIBUTE BY logs.error_seq" can be used if it would result better distribution.
After writing to here, I've found it's not so useful. Should I remove it?
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13423825#comment-13423825 ]
Namit Jain commented on HIVE-3286:
----------------------------------
But, if you know the skewed keys, cant you create a group for each of the skewed key ?
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13424312#comment-13424312 ]
Namit Jain commented on HIVE-3286:
----------------------------------
I am sorry, I was not clear, what I meant was the following:
If you know keys 10, 25 and 40 are skewed, accounting for nearly 5%, 6% and 7% of data respectively,
can't you issue the following
select count(*) from src a join src b on a.key=b.key skew on (
a.key = '10' CLUSTER BY 5 PERCENT,
a.key = '25' CLUSTER BY 6 PERCENT,
a.key = '40' CLUSTER BY 7 PERCENT);
I am not clear on why do you need DISTRIBUTE BY ?
KEYS and EXPRESSIONS should lead to the same distribution.
Isn't that right ?
I am sorry, can you give a clear example of where you see the benefit of using DISTRIBUTE BY ?
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Namit Jain updated HIVE-3286:
-----------------------------
Status: Open (was: Patch Available)
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13419040#comment-13419040 ]
Navis commented on HIVE-3286:
-----------------------------
The idea of this issue was conceived couple of months ago and I've seen the document later. I love the systemic approach in it.
I've considered using that but decided to implement this cause this seemed to allow more freedom in schema(without list bucketing).
If this is not appropriate for hive, I can keep this only for internal use.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13420008#comment-13420008 ]
Namit Jain commented on HIVE-3286:
----------------------------------
@Navis, can you explain the semantics of the above grammar ?
What doe SKEWED BY, DISTRIBUTE BY imply ?
Also, in the base case:
select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
are you expecting skewed keys for key <= 49.
Is it true that the skewed keys will only be handled by reducers ?
If yes, why would it reduce the execution time ? The main advantage should be that reducer wont get any other key, so
wont be burdened. Is that the idea ?
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13419027#comment-13419027 ]
Namit Jain commented on HIVE-3286:
----------------------------------
Navis, Nadeem is already working on this in a different approach
https://cwiki.apache.org/Hive/skewed-join-optimization.html
I am not sure if there is a jira, but I know he is pretty close to getting one out.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Namit Jain updated HIVE-3286:
-----------------------------
Status: Open (was: Patch Available)
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Commented] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Namit Jain (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13423734#comment-13423734 ]
Namit Jain commented on HIVE-3286:
----------------------------------
// distributes randomly, disperses non driving aliases to all partitions in the skew group
public static final int SKEW_RULE_RANDOM = 0;
Why is this needed ?
I mean, wont it be very expensive ?
2. KEYS : determined by hash value of keys (same with previous)
3. expression : determined by hash of object evaluated by user-provided expression
Wont the above 2 always lead to the same expression ?
Basically, why is distribute by needed at all ? Cant we always use the KEYS semantics ?
This seems too confusing.
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira
[jira] [Updated] (HIVE-3286) Explicit skew join on user provided
condition
Posted by "Navis (JIRA)" <ji...@apache.org>.
[ https://issues.apache.org/jira/browse/HIVE-3286?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Navis updated HIVE-3286:
------------------------
Status: Patch Available (was: Open)
> Explicit skew join on user provided condition
> ---------------------------------------------
>
> Key: HIVE-3286
> URL: https://issues.apache.org/jira/browse/HIVE-3286
> Project: Hive
> Issue Type: Improvement
> Components: Query Processor
> Affects Versions: 0.10.0
> Reporter: Navis
> Assignee: Navis
> Priority: Minor
>
> Join operation on table with skewed data takes most of execution time handling the skewed keys. But mostly we already know about that and even know what is look like the skewed keys.
> If we can explicitly assign reducer slots for the skewed keys, total execution time could be greatly shortened.
> As for a start, I've extended join grammar something like this.
> {code}
> select * from src a join src b on a.key=b.key skew on (a.key+1 < 50, a.key+1 < 100, a.key < 150);
> {code}
> which means if above query is executed by 20 reducers, one reducer for a.key+1 < 50, one reducer for 50 <= a.key+1 < 100, one reducer for 99 <= a.key < 150, and 17 reducers for others (could be extended to assign more than one reducer later)
> This can be only used with common-inner-equi joins. And skew condition should be composed of join keys only.
> Work till done now will be updated shortly after code cleanup.
> ----------------------------
> Skew expressions* in "SKEW ON (expr, expr, ...)" are evaluated sequentially at runtime, and first 'true' one decides skew group for the row. Each skew group has reserved partition slot(s), to which all rows in a group would be assigned.
> The number of partition slot reserved for each group is decided also at runtime by simple calculation of percentage. If a skew group is "CLUSTER BY 20 PERCENT" and total partition slot (=number of reducer) is 20, that group will reserve 4 partition slots, etc.
> "DISTRIBUTE BY" decides how the rows in a group is dispersed in the range of reserved slots (If there is only one slot for a group, this is meaningless). Currently, three distribution policies are available: RANDOM, KEYS, <expression>.
> 1. RANDOM : rows of driver** alias are dispersed by random and rows of non-driver alias are duplicated for all the slots (default if not specified)
> 2. KEYS : determined by hash value of keys (same with previous)
> 3. expression : determined by hash of object evaluated by user-provided expression
> Only possible with inner, equi, common-joins. Not yet supports join tree merging.
> Might be used by other RS users like "SORT BY" or "GROUP BY"
> If there exists column statistics for the key, it could be possible to apply automatically.
> For example, if 20 reducers are used for the query below,
> {code}
> select count(*) from src a join src b on a.key=b.key skew on (
> a.key = '0' CLUSTER BY 10 PERCENT,
> b.key < '100' CLUSTER BY 20 PERCENT DISTRIBUTE BY upper(b.key),
> cast(a.key as int) > 300 CLUSTER BY 40 PERCENT DISTRIBUTE BY KEYS);
> {code}
> group-0 will reserve slots 6~7, group-1 8~11, group-2 12~19 and others will reserve slots 0~5.
> For a row with key='0' from alias a, the row is randomly assigned in the range of 6~7 (driver alias) : 6 or 7
> For a row with key='0' from alias b, the row is disributed for all slots in 6~7 (non-driver alias) : 6 and 7
> For a row with key='50', the row is assigned in the range of 8~11 by hashcode of upper(b.key) : 8 + (hash(upper(key)) % 4)
> For a row with key='500', the row is assigned in the range of 12~19 by hashcode of join key : 12 + (hash(key) % 8)
> For a row with key='200', this is not belong to any skew group : hash(key) % 6
> *expressions in skew condition :
> 1. all expressions should be made of expression in join condition, which means if join condition is "a.key=b.key", user can make any expression with "a.key" or "b.key". But if join condition is a.key+1=b.key, user cannot make expression with "a.key" solely (should make expression with "a.key+1").
> 2. all expressions should reference one and only-one side of aliases. For example, simple constant expressions or expressions referencing both side of join condition ("a.key+b.key<100") is not allowed.
> 3. all functions in expression should be deteministic and stateless.
> 4. if "DISTRIBUTED BY expression" is used, distibution expression also should have same alias with skew expression.
> **driver alias :
> 1. driver alias means the sole referenced alias from skew expression, which is important for RANDOM distribution. rows of driver alias are assigned to single slot randomly, but rows of non-driver alias are duplicated for all the slots. So, driver alias should be the biggest one in join aliases.
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira