Original code is a bit huge and has lots of joins and aggregation so just attaching a small snippet

Sample code

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from multiprocessing.pool import ThreadPool
import pyspark
class do_multithreading:
    def run_multithreader (data):
         spark = data[0]
         df = data[1]
         filter_val = data[3]
         # I am repartitioning and clustering the data along with persist to achieve 
         # Bucket join effect when joining.
         
         df = df.filter(col('validation_Date') == filter_val)\
              .repartition(500,'validation_Date','col_x')
              .sortWithinPartitions( 'col_xx','col_yy' )\
              .persist( pyspark.StorageLevel.MEMORY_ONLY_2)
         
         df=df.groupBy( 'col_1','col_2','validation_Date' ).agg({'col3':'sum'})
         df=df.select( 'col_1','col_2',
                 col('sum(col_3)').alias('sum_of_col3'),'validation_Date')
         df2=spark.read.table('dim_db.some_dimesion')
         df = df.join(broadcast(df2),how='left',on=['validation_date','col1'])
         df.coalesce(10).write.mode('append')\
                 .option("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")\      
                 .partitionBy('validation_Date').saveAsTable('db.target_table')

def main():
    spark=SparkSession.builder.appName('Multithreading Pyspark')\
                      .enableHiveSupport().getOrCreate()
    obj=do_multithreading()
    df=spark.read.table('db.tb1')
    pools=[[spark,df,'2021-08-21'],[spark,df,'2021-08-22'],[spark,df,'2021-08-23']]
    tpool = ThreadPool(processes=len(pools))
    tpool.map(obj.run_multithreader,pools)

What this solved:

Using multithreading with spark along with FAIR pool schedular helped achieve job execution time reduction from 4 hrs to 1 hr but In spark I am seeing some issue with job failure due to concurrent write to same partitioned table.This issue occur's like once in 2 week or some time a month.

LOGS:

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: ShuffleMapStage 1866 (persist at :0) has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 1613 at org.apache.spark.MapOutputTracker$$anonfun$convertMapStatuses$2.apply(MapOutputTracker.scala:867) at org.apache.spark.MapOutputTracker$$anonfun$convertMapStatuses$2.apply(MapOutputTracker.scala:863) at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186) at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732) at org.apache.spark.MapOutputTracker$.convertMapStatuses(MapOutputTracker.scala:863) at org.apache.spark.MapOutputTrackerWorker.getMapSizesByExecutorId(MapOutputTracker.scala:677) at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:49) at org.apache.spark.sql.execution.ShuffledRowRDD.compute(ShuffledRowRDD.scala:165) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324) at org.apache.spark.rdd.RDD.iterator(RDD.scala:288) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:49) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324) at org.apache.spark.rdd.RDD.iterator(RDD.scala:288) at org.apache.spark.rdd.UnionRDD.compute(UnionRDD.scala:105) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324) at org.apache.spark.rdd.RDD.iterator(RDD.scala:288) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:49) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324) at org.apache.spark.rdd.RDD.iterator(RDD.scala:288) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53) at org.apache.spark.scheduler.Task.run(Task.scala:109) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)