[SPARK-8300] DataFrame hint for broadcast join.

[rxin]

Users can now do

left.join(broadcast(right), "joinKey")

to give the query planner a hint that "right" DataFrame is small and should be broadcasted.

[patmcdonough]

@rxin - This is great. I'm a fan of the alternative syntax we chatted about:

left.join(right.broadcast, "joinKey")

[SparkQA]

Test build #34641 has finished for PR 6751 at commit 3ad8de6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BroadcastHint(child: LogicalPlan) extends UnaryNode

[marmbrus]

Could we just have this be StatisticsHint and override the statistics? I'm afraid that we are going to forget to add cases in the future as we broadcast more type of joins.

[rxin]

that sounds good.

[marmbrus]

I'm thinking about this more and actually maybe what we want to do is have this specific node, and a pattern that recognizes canBroadcast. This pattern can check either for a small enough size or this hint and we can use that anywhere we are planning a broadcast operator. The reasoning is that messing with statistics could have other consequences.

[chenghao-intel]

+1
Add specific node is also doable, probably the node can be named like Hint in LogicalPlan, and could be more than broadcast, like uniq_key etc.

Any idea?

[marmbrus]

Regarding broadcast(...) vs .broadcast, I don't really care too much. However, I think we need to come up with a convention. For example, I was recently confused that its not isNotNull(column) but instead column.isNotNull.

[rxin]

We should add isNotNull too.

The isNotNull was from SchemaRDD.

[marmbrus]

I'm confused now: both column.isNotNull and column.isNull work today. I was expecting them to be functions instead of being functions that column defines. I'm mostly wondering how we decide what goes where.

[rxin]

I meant adding isNotNull as a function, rather than a member method of Column.

Basically most of those Column.function were inherited from SchemaRDD time.

[rxin]

@marmbrus is this what you had in mind?

[marmbrus]

Yep!

[SparkQA]

Test build #35520 has finished for PR 6751 at commit 953eec2.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[marmbrus]

Merging to master!

[sridharsubramanian62]

How can this hint be used with Spark-SQL ?

Below example does not seem to work.

select a.col1,b.col2 from tableA a left outer join broadcast(tableB) on a.col1 = b.col1

[marmbrus]

It can only be used from the dataframe API.
On Oct 8, 2015 3:20 AM, "sridharsubramanian62" notifications@github.com
wrote:

How can this hint be used with Spark-SQL ?

Below example does not seem to work.

select a.col1,b.col2 from tableA a left outer join broadcast(tableB) on
a.col1 = b.col1

—
Reply to this email directly or view it on GitHub
#6751 (comment).

[RoiViber]

Using broadcast(...) results in "java.lang.AssertionError: assertion failed: No plan for BroadcastHint" as in https://issues.apache.org/jira/browse/SPARK-12275 so this solution is not working most of the time

[fjh100456]

@rxin @marmbrus
Is there another way to broadcast table with the spark-sql now, except by spark.sql.autoBroadcastJoinThreshold?
And if no, is it a good way to broadcast table by user conf,such as spark.sql.autoBroadcastJoin.mytable=true to broadcast a table named mytable.

[rxin]

Isn't the hint available in SQL?

[sridharsubramanian62]

Its available from spark 2.2.0. On Tuesday, October 10, 2017, 1:46:27 PM PDT, Reynold Xin <notifications@github.com> wrote: Isn't the hint available in SQL? — You are receiving this because you commented. Reply to this email directly, view it on GitHub, or mute the thread.

[fjh100456]

Is there an example?
I use broadcast like the following, but it perform an error.Would you be so kind as to show me an example?

spark-sql> select a.* from tableA a left outer join broadcast(tableB) b on a.a=b.a;
17/10/11 09:06:40 INFO HiveMetaStore: 0: get_table : db=default tbl=tablea
17/10/11 09:06:40 INFO audit: ugi=root ip=unknown-ip-addr cmd=get_table : db=default tbl=tablea

Error in query: cannot resolve 'tableB' given input columns: []; line 1 pos 51;

'Project [ArrayBuffer(a).*]
+- 'Join LeftOuter, ('a.a = 'b.a)
:- SubqueryAlias a
: +- SubqueryAlias tablea
: +- HiveTableRelation default.tablea, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [a#94, b#95, c#96]
+- 'SubqueryAlias b
+- 'UnresolvedTableValuedFunction broadcast, ['tableB]

[sridharsubramanian62]

Let me know if this does not help [SPARK-16475] Broadcast Hint for SQL Queries - ASF JIRA | | | | [SPARK-16475] Broadcast Hint for SQL Queries - ASF JIRA | | | On Tuesday, October 10, 2017, 6:25:57 PM PDT, fjh100456 <notifications@github.com> wrote: Is there an example? I use broadcast like the following, but it perform an error.Would you be so kind as to show me an example? spark-sql> select a.* from tableA a left outer join broadcast(tableB) b on a.a=b.a; 17/10/11 09:06:40 INFO HiveMetaStore: 0: get_table : db=default tbl=tablea 17/10/11 09:06:40 INFO audit: ugi=root ip=unknown-ip-addr cmd=get_table : db=default tbl=tablea Error in query: cannot resolve 'tableB' given input columns: []; line 1 pos 51; 'Project [ArrayBuffer(a).*] +- 'Join LeftOuter, ('a.a = 'b.a) :- SubqueryAlias a : +- SubqueryAlias tablea : +- HiveTableRelation default.tablea, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [a#94, b#95, c#96] +- 'SubqueryAlias b +- 'UnresolvedTableValuedFunction broadcast, ['tableB] — You are receiving this because you commented. Reply to this email directly, view it on GitHub, or mute the thread.

[fjh100456]

With /*+ broadcast(table) */, it works well, thank you very much.