I created skewed data to test a salting approach and tried three different solutions, but none achieved the desired results with a significant runtime improvement. Can you guide me on the best approach to solve this problem effectively?
import pyspark.sql.functions as F
df1 = spark.range(300_000_000).withColumn('value',F.when(F.rand() < 0.6,1).otherwise((F.rand() * 100).cast('int)).drip('id')
df2 = spark.range(200_000_000).withColumn('value',F.when(F.rand() < 0.2,4).otherwise((F.rand() * 100).cast('int)).drip('id')
final_df = df1.join(df2,on='value',how='inner')
finaldf.write.format('parquet).save(path)
Process 1
Enabled AQE and other settings like skewjoin,enabled=true,coalescePartitions.enabled=True
and shuffle.partition.=auto
It was running for more than 20 minutes, and I cancelled the job manually.
Process 2
Salting techniques. Disabled AQE and added shuffle partition size = 1000
df1 = (df1.withColumn('salt_numbers',F.expr('sequence(0,1)'))
.withColum('salt',F.explode('salt_numbers'))
.drop('salt_numbers))
df2 = (df2.withColumn('salt_numbers',F.expr('sequence(0,3)'))
.withColum('salt',F.explode('salt_numbers'))
.drop('salt_numbers))
Now both datasets have equal size (600_000_00) and added joining column - salt. on=['value','salt'] . It ran more that 30min and cancelled the job manually.
Process 3
Salting techniques: Enabled AQE
df1 = (df1.withColumn('salt_numbers',F.expr('sequence(0,1)'))
.withColum('salt',F.explode('salt_numbers'))
.drop('salt_numbers))
df2 = (df2.withColumn('salt_numbers',F.expr('sequence(0,3)'))
.withColum('salt',F.explode('salt_numbers'))
.drop('salt_numbers))
Now both datasets have equal size (600_000_00) and added joining column - salt. on=['value','salt'] . also took similar 30 min and manually cancelled the job.
Note: Databricks cluster size is 32 gb 4 cores and 8 workers. Please share your ideas on how efficiently we can run the job.