[
https://issues.apache.org/jira/browse/SPARK-15219?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yi Zhou updated SPARK-15219:
----------------------------
Description:
We observed an interesting thing about broadcast Hash join( similar to Map Join
in Hive) when comparing the implementation by Hive on MR engine. The blew query
is a multi-way join operation based on 3 tables including product_reviews, 2
run-time temporary result tables(fsr and fwr) from ‘select’ query operation and
also there is a two-way join(1 table and 1 run-time temporary table) in both
'fsr' and 'fwr',which cause slower performance than Hive on MR. We investigated
the difference between Spark SQL and Hive on MR engine and found that there are
total of 5 map join tasks with tuned map join parameters in Hive on MR but
there are only 2 broadcast hash join tasks in Spark SQL even if we set a larger
threshold(e.g.,1GB) for broadcast hash join. From our investigation, it seems
that if there is run-time temporary table in join operation in Spark SQL engine
it will not detect such table for enabling broadcast hash join optimization.
Core SQL snippet:
{code}
INSERT INTO TABLE q19_spark_sql_power_test_0_result
SELECT *
FROM
( --wrap in additional FROM(), because Sorting/distribute by with UDTF in
select clause is not allowed
SELECT extract_sentiment(pr.pr_item_sk, pr.pr_review_content) AS
(
item_sk,
review_sentence,
sentiment,
sentiment_word
)
FROM product_reviews pr,
(
--store returns in week ending given date
SELECT sr_item_sk, SUM(sr_return_quantity) sr_item_qty
FROM store_returns sr,
(
-- within the week ending a given date
SELECT d1.d_date_sk
FROM date_dim d1, date_dim d2
WHERE d1.d_week_seq = d2.d_week_seq
AND d2.d_date IN ( '2004-03-8' ,'2004-08-02' ,'2004-11-15', '2004-12-20' )
) sr_dateFilter
WHERE sr.sr_returned_date_sk = d_date_sk
GROUP BY sr_item_sk --across all store and web channels
HAVING sr_item_qty > 0
) fsr,
(
--web returns in week ending given date
SELECT wr_item_sk, SUM(wr_return_quantity) wr_item_qty
FROM web_returns wr,
(
-- within the week ending a given date
SELECT d1.d_date_sk
FROM date_dim d1, date_dim d2
WHERE d1.d_week_seq = d2.d_week_seq
AND d2.d_date IN ( '2004-03-8' ,'2004-08-02' ,'2004-11-15', '2004-12-20' )
) wr_dateFilter
WHERE wr.wr_returned_date_sk = d_date_sk
GROUP BY wr_item_sk --across all store and web channels
HAVING wr_item_qty > 0
) fwr
WHERE fsr.sr_item_sk = fwr.wr_item_sk
AND pr.pr_item_sk = fsr.sr_item_sk --extract product_reviews for found items
-- equivalent across all store and web channels (within a tolerance of +/-
10%)
AND abs( (sr_item_qty-wr_item_qty)/ ((sr_item_qty+wr_item_qty)/2)) <= 0.1
)extractedSentiments
WHERE sentiment= 'NEG' --if there are any major negative reviews.
ORDER BY item_sk,review_sentence,sentiment,sentiment_word
;
{code}
was:
We observed an interesting thing about broadcast Hash join( similar to Map Join
in Hive) when comparing the implementation by Hive on MR engine. The blew query
is a multi-way join operation based on 3 tables including product_reviews, 2
run-time temporary result tables(fsr and fwr) from ‘select’ query operation and
also there is a two-way join(1 table and 1 run-time temporary table) in both
'fsr' and 'fwr'. We investigated the difference between Spark SQL and Hive on
MR engine and found that there are total of 5 map join tasks with tuned map
join parameters in Hive on MR but there are only 2 broadcast hash join tasks in
Spark SQL even if we set a larger threshold(e.g.,1GB) for broadcast hash join.
From our investigation, it seems that if there is run-time temporary table in
join operation in Spark SQL engine it will not detect such table for enabling
broadcast hash join optimization.
Core SQL snippet:
{code}
INSERT INTO TABLE q19_spark_sql_power_test_0_result
SELECT *
FROM
( --wrap in additional FROM(), because Sorting/distribute by with UDTF in
select clause is not allowed
SELECT extract_sentiment(pr.pr_item_sk, pr.pr_review_content) AS
(
item_sk,
review_sentence,
sentiment,
sentiment_word
)
FROM product_reviews pr,
(
--store returns in week ending given date
SELECT sr_item_sk, SUM(sr_return_quantity) sr_item_qty
FROM store_returns sr,
(
-- within the week ending a given date
SELECT d1.d_date_sk
FROM date_dim d1, date_dim d2
WHERE d1.d_week_seq = d2.d_week_seq
AND d2.d_date IN ( '2004-03-8' ,'2004-08-02' ,'2004-11-15', '2004-12-20' )
) sr_dateFilter
WHERE sr.sr_returned_date_sk = d_date_sk
GROUP BY sr_item_sk --across all store and web channels
HAVING sr_item_qty > 0
) fsr,
(
--web returns in week ending given date
SELECT wr_item_sk, SUM(wr_return_quantity) wr_item_qty
FROM web_returns wr,
(
-- within the week ending a given date
SELECT d1.d_date_sk
FROM date_dim d1, date_dim d2
WHERE d1.d_week_seq = d2.d_week_seq
AND d2.d_date IN ( '2004-03-8' ,'2004-08-02' ,'2004-11-15', '2004-12-20' )
) wr_dateFilter
WHERE wr.wr_returned_date_sk = d_date_sk
GROUP BY wr_item_sk --across all store and web channels
HAVING wr_item_qty > 0
) fwr
WHERE fsr.sr_item_sk = fwr.wr_item_sk
AND pr.pr_item_sk = fsr.sr_item_sk --extract product_reviews for found items
-- equivalent across all store and web channels (within a tolerance of +/-
10%)
AND abs( (sr_item_qty-wr_item_qty)/ ((sr_item_qty+wr_item_qty)/2)) <= 0.1
)extractedSentiments
WHERE sentiment= 'NEG' --if there are any major negative reviews.
ORDER BY item_sk,review_sentence,sentiment,sentiment_word
;
{code}
> [Spark SQL] it don't support to detect runtime temporary table for enabling
> broadcast hash join optimization
> ------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-15219
> URL: https://issues.apache.org/jira/browse/SPARK-15219
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Reporter: Yi Zhou
>
> We observed an interesting thing about broadcast Hash join( similar to Map
> Join in Hive) when comparing the implementation by Hive on MR engine. The
> blew query is a multi-way join operation based on 3 tables including
> product_reviews, 2 run-time temporary result tables(fsr and fwr) from
> ‘select’ query operation and also there is a two-way join(1 table and 1
> run-time temporary table) in both 'fsr' and 'fwr',which cause slower
> performance than Hive on MR. We investigated the difference between Spark SQL
> and Hive on MR engine and found that there are total of 5 map join tasks with
> tuned map join parameters in Hive on MR but there are only 2 broadcast hash
> join tasks in Spark SQL even if we set a larger threshold(e.g.,1GB) for
> broadcast hash join. From our investigation, it seems that if there is
> run-time temporary table in join operation in Spark SQL engine it will not
> detect such table for enabling broadcast hash join optimization.
> Core SQL snippet:
> {code}
> INSERT INTO TABLE q19_spark_sql_power_test_0_result
> SELECT *
> FROM
> ( --wrap in additional FROM(), because Sorting/distribute by with UDTF in
> select clause is not allowed
> SELECT extract_sentiment(pr.pr_item_sk, pr.pr_review_content) AS
> (
> item_sk,
> review_sentence,
> sentiment,
> sentiment_word
> )
> FROM product_reviews pr,
> (
> --store returns in week ending given date
> SELECT sr_item_sk, SUM(sr_return_quantity) sr_item_qty
> FROM store_returns sr,
> (
> -- within the week ending a given date
> SELECT d1.d_date_sk
> FROM date_dim d1, date_dim d2
> WHERE d1.d_week_seq = d2.d_week_seq
> AND d2.d_date IN ( '2004-03-8' ,'2004-08-02' ,'2004-11-15',
> '2004-12-20' )
> ) sr_dateFilter
> WHERE sr.sr_returned_date_sk = d_date_sk
> GROUP BY sr_item_sk --across all store and web channels
> HAVING sr_item_qty > 0
> ) fsr,
> (
> --web returns in week ending given date
> SELECT wr_item_sk, SUM(wr_return_quantity) wr_item_qty
> FROM web_returns wr,
> (
> -- within the week ending a given date
> SELECT d1.d_date_sk
> FROM date_dim d1, date_dim d2
> WHERE d1.d_week_seq = d2.d_week_seq
> AND d2.d_date IN ( '2004-03-8' ,'2004-08-02' ,'2004-11-15',
> '2004-12-20' )
> ) wr_dateFilter
> WHERE wr.wr_returned_date_sk = d_date_sk
> GROUP BY wr_item_sk --across all store and web channels
> HAVING wr_item_qty > 0
> ) fwr
> WHERE fsr.sr_item_sk = fwr.wr_item_sk
> AND pr.pr_item_sk = fsr.sr_item_sk --extract product_reviews for found items
> -- equivalent across all store and web channels (within a tolerance of +/-
> 10%)
> AND abs( (sr_item_qty-wr_item_qty)/ ((sr_item_qty+wr_item_qty)/2)) <= 0.1
> )extractedSentiments
> WHERE sentiment= 'NEG' --if there are any major negative reviews.
> ORDER BY item_sk,review_sentence,sentiment,sentiment_word
> ;
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
