There are circumstances when tasks (Spark action, e.g. save, count, etc) in a PySpark job can be spawned on separate threads. Doing so, optimizes distribution of tasks on executor cores. From , Spark docs If the jobs at the head of the queue don’t need to use the whole cluster, later jobs can start to run right away, but if the jobs at the head of the queue are large, then later jobs may be delayed significantly. By default, Spark’s scheduler runs jobs in FIFO fashion. Each job is divided into “stages” (e.g. map and reduce phases), and the first job gets priority on all available resources while its stages have tasks to launch, then the second job gets priority, etc. Threading tasks addresses this particular situation: through, if the jobs at the head of the queue are large, then later jobs may be delayed significantly, Spark assigns tasks between jobs in a “round robin” fashion, so that all jobs get a roughly equal share of cluster resources. This means that short jobs submitted while a long job is running can start receiving resources right away and still get good response times, without waiting for the long job to finish. Simple But Very Common Use Case Saving multiple results to parquet files is a common action. Here’s how to thread three action. write.parquet An important reminder is to set set('spark.scheduler.mode','FAIR') in the sparkContext. threading conf = SparkConf().\
            setAppName(app_name).\
            set( , ).\
            set( , ).\
            set( , ) sc = SparkContext(conf=conf) :
        sc._jvm.org.apache.hadoop.hive.conf.HiveConf()
        sqlCtx = sqlContext = HiveContext(sc) py4j.protocol.Py4JError:
        sqlCtx = sqlContext = SQLContext(sc) sc, sqlContext sc.setLocalProperty( , str(pool_id))
    df.repartition( ).write.parquet(path)
    sc.setLocalProperty( , )
sc, sqlContext = set_spark_context( ) path1 = path2 = path3 = df1_write = threading.Thread(target=saveParquet, args=(df1,path1, ,sc,))
df2_write = threading.Thread(target=saveParquet, args=(df2,path2, ,sc,))
df3_write = threading.Thread(target=saveParquet, args=(df3,path3, ,sc,))
df1.start()
df2.start()
df3.start()
df1.join()
df2.join()
df3.join() import : def set_spark_context (app_name=None) 'spark.hadoop.mapreduce.output.fileoutputformat.compress' 'false' 'spark.sql.parquet.compression.codec' 'uncompressed' 'spark.scheduler.mode' 'FAIR' # need to set this on the sparkContext try except return #function to be called on spawned threads* : def saveParquet (df,path,pool_id,sc) "spark.scheduler.pool" 1 "spark.scheduler.pool" None "Simple TASK" # Do processing here to produce three dataframes to be written as parquets. # The tree results are df1, df2, df3 "/path/1" "/path/2" "/path/3" 1 2 3 When this job is submitted on the cluster, at first it will show 1 scheduler pool. However, at the stage where the actual snippet above is already being interpreted, it will show the 3 pools spawned and the active pools currently being processed. Hopefully, this is something useful for us trying to speed up running jobs.

Apache

Target

How-to Declutter Your Data Science Workspace

Serve Data Models with MLFlow in Production

I do data science and engineering with PySpark, Python & Open Source tools.

Threaded Tasks in PySpark Jobs

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

Data Product Managers and the Data Mesh

3 Best Hadoop Alternatives to Consider for Migration

5 Big Data Problems and How to Solve Them

6 Database Migration Tools For Complete Data Integrity & More

7 Data Analytics Tools to Boost Your Business

Data Product Managers and the Data Mesh

3 Best Hadoop Alternatives to Consider for Migration

5 Big Data Problems and How to Solve Them

6 Database Migration Tools For Complete Data Integrity & More

7 Data Analytics Tools to Boost Your Business

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps