Fast Spark alternative to WHERE column IN other_column

user_9019559 0
Viewed 494

I am looking for a fast PySpark alternative to

SELECT foo FROM bar
WHERE foo IN (SELECT baz FROM bar)

Collecting beforehand into a Python list is absolutely not an option as the dataframes handled are quite large and collect takes up tremendous amounts of time relative to the other option I came up with. So I can't think of a way to use a native PySparkian where(col(bar).isin(baz)) since baz would have to be a list in this case.

One option I came up with is right JOIN as a replacement of IN and left_semi JOIN as a replacement of NOT IN, consider a following example: 

bar_where_foo_is_in_baz = bar.join(bar.select('baz').alias('baz_'), col('foo') == col('baz_'), 'right').drop('baz_')

This however is quite verbose, pretty hard to interpret when read after a while and results in a fair bit of head scratching when a larger number of conditions are handled within WHERE so I'd like to avoid that.

Are there any other options?

EDIT (please read):

As I have seem to have misled quite a number of answers, my specific requirement is translating a "WHERE - IN" clause into a PySpark without .collect() or in general, mapping to Pythonic list (as the internal function .isin() would require me).

pyspark where-in sql
1 Answers

From Jacek Laskowski's GitBook

Spark SQL uses broadcast join (aka broadcast hash join) instead of hash join to optimize join queries when the size of one side data is below spark.sql.autoBroadcastJoinThreshold

Broadcast join can be very efficient for joins between a large table (fact) with relatively small tables (dimensions) that could then be used to perform a star-schema join. It can avoid sending all data of the large table over the network.

And also

Spark SQL 2.2 supports BROADCAST hints using broadcast standard function or SQL comments:

  • SELECT /*+ MAPJOIN(b) */ …​
  • SELECT /*+ BROADCASTJOIN(b) */ …​
  • SELECT /*+ BROADCAST(b) */ …​

Actually the Spark documentation is more comprehensive about the hints:

The BROADCAST hint guides Spark to broadcast each specified table when joining them with another table or view. When Spark deciding the join methods, the broadcast hash join (i.e., BHJ) is preferred, even if the statistics is above the configuration spark.sql.autoBroadcastJoinThreshold.

When both sides of a join are specified, Spark broadcasts the one having the lower statistics. Note Spark does not guarantee BHJ is always chosen, since not all cases (e.g. full outer join) support BHJ.

Finally, if you are serious about developping efficient Spark jobs, you should spend some time on understanding how that beast works.

For instance that presentation about joins from Databricks should be helpful

user_51623720
Related