What is an optimized way of joining large tables in Spark SQL

by

My default advice on how to optimize joins is:

  1. Use a broadcast join if you can (see this notebook). From your question it seems your tables are large and a broadcast join is not an option.

  2. Consider using a very large cluster (it’s cheaper that you may think). $250 right now (6/2016) buys about 24 hours of 800 cores with 6Tb RAM and many SSDs on the EC2 spot instance market. When thinking about total cost of a big data solution, I find that humans tend to substantially undervalue their time.

  3. Use the same partitioner. See this question for information on co-grouped joins.

  4. If the data is huge and/or your clusters cannot grow such that even (3) above leads to OOM, use a two-pass approach. First, re-partition the data and persist using partitioned tables (dataframe.write.partitionBy()). Then, join sub-partitions serially in a loop, “appending” to the same final result table.

Side note: I say “appending” above because in production I never use SaveMode.Append. It is not idempotent and that’s a dangerous thing. I use SaveMode.Overwrite deep into the subtree of a partitioned table tree structure. Prior to 2.0.0 and 1.6.2 you’ll have to delete _SUCCESS or metadata files or dynamic partition discovery will choke.

Hope this helps.

More Related Contents:

  1. Using a column value as a parameter to a spark DataFrame function
  2. Split Spark Dataframe string column into multiple columns
  3. DataFrame join optimization – Broadcast Hash Join
  4. Overwrite specific partitions in spark dataframe write method
  5. How to control partition size in Spark SQL
  6. Avoid performance impact of a single partition mode in Spark window functions
  7. How to import multiple csv files in a single load?
  8. Convert date from String to Date format in Dataframes
  9. Spark DataFrame: count distinct values of every column
  10. How to split a list to multiple columns in Pyspark?
  11. pyspark dataframe filter or include based on list
  12. Use collect_list and collect_set in Spark SQL
  13. What is the maximum size for a broadcast object in Spark?
  14. How to find mean of grouped Vector columns in Spark SQL?
  15. Generate a Spark StructType / Schema from a case class
  16. How to loop through each row of dataFrame in pyspark
  17. How does createOrReplaceTempView work in Spark?
  18. Fill in null with previously known good value with pyspark
  19. How to group by common element in array?
  20. PySpark – get row number for each row in a group
  21. Keep only duplicates from a DataFrame regarding some field
  22. Using windowing functions in Spark
  23. How can I access python variable in Spark SQL?
  24. reduce result datasets into single dataset
  25. How to exclude multiple columns in Spark dataframe in Python
  26. Stratified sampling with pyspark
  27. How to use groupBy to collect rows into a map?
  28. Convert between spark.SQL DataFrame and pandas DataFrame [duplicate]
  29. How to turn off scientific notation in pyspark?
  30. When to use Spark DataFrame/Dataset API and when to use plain RDD?

Leave a Comment