Efficient pyspark join

by

you can also use a two-pass approach, in case it suits your requirement.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.
It was nicely explained by Sim. see link below

two pass approach to join big dataframes in pyspark

based on case explained above I was able to join sub-partitions serially in a loop and then persisting joined data to hive table.

Here is the code.

from pyspark.sql.functions import *
emp_df_1.withColumn("par_id",col('emp_id')%5).repartition(5, 'par_id').write.format('orc').partitionBy("par_id").saveAsTable("UDB.temptable_1")
emp_df_2.withColumn("par_id",col('emp_id')%5).repartition(5, 'par_id').write.format('orc').partitionBy("par_id").saveAsTable("UDB.temptable_2")

So, if you are joining on an integer emp_id, you can partition by the ID modulo some number and this way you can re distribute the load across the spark partitions and records having similar keys will be grouped together and reside on same partition.
you can then read and loop through each sub partition data and join both the dataframes and persist them together.

counter =0;
paritioncount = 4;
while counter<=paritioncount:
    query1 ="SELECT * FROM UDB.temptable_1 where par_id={}".format(counter)
    query2 ="SELECT * FROM UDB.temptable_2 where par_id={}".format(counter)
    EMP_DF1 =spark.sql(query1)
    EMP_DF2 =spark.sql(query2)
    df1 = EMP_DF1.alias('df1')
    df2 = EMP_DF2.alias('df2')
    innerjoin_EMP = df1.join(df2, df1.emp_id == df2.emp_id,'inner').select('df1.*')
    innerjoin_EMP.show()
    innerjoin_EMP.write.format('orc').insertInto("UDB.temptable")
    counter = counter +1

I have tried this and this is working fine. This is just an example to demo the two-pass approach. your join conditions may vary and the number of partitions also depending on your data size.

More Related Contents:

  1. Finding duplicates from large data set using Apache Spark
  2. Using a column value as a parameter to a spark DataFrame function
  3. Find maximum row per group in Spark DataFrame
  4. How do I split an RDD into two or more RDDs?
  5. java.lang.IllegalArgumentException at org.apache.xbean.asm5.ClassReader.(Unknown Source) with Java 10
  6. Split Spark Dataframe string column into multiple columns
  7. How to access element of a VectorUDT column in a Spark DataFrame?
  8. Python Spark Cumulative Sum by Group Using DataFrame
  9. How to split a list to multiple columns in Pyspark?
  10. pyspark dataframe filter or include based on list
  11. Link Spark with iPython Notebook
  12. How to save/insert each DStream into a permanent table
  13. Spark load data and add filename as dataframe column
  14. PySpark: how to resample frequencies
  15. Fill in null with previously known good value with pyspark
  16. Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?
  17. PySpark – get row number for each row in a group
  18. Save ML model for future usage
  19. Apache spark dealing with case statements
  20. Keep only duplicates from a DataFrame regarding some field
  21. reading json file in pyspark
  22. pyspark: count distinct over a window
  23. Serialize a custom transformer using python to be used within a Pyspark ML pipeline
  24. Not able to cat dbfs file in databricks community edition cluster. FileNotFoundError: [Errno 2] No such file or directory:
  25. PySpark: How to fillna values in dataframe for specific columns?
  26. pyspark: rolling average using timeseries data
  27. Spark ALS predictAll returns empty
  28. Why is the fold action necessary in Spark?
  29. How to add a SparkListener from pySpark in Python?
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment