Python using multiprocessing on PostgreSQL queries reduce runtime

Question

I have the following code in Python, which will execute 5 queries in a row. The average runtime per query is around 181.1 seconds (~3 minutes), and the total runtime for all 5 queries is 905.4 seconds (~15 minutes). Eventually, after loading the data into DataFrames, I am going to perform ETL work (mostly looking for errors, data quality issues, and inconsistencies), but before that I want to try to leverage multiprocessing to shave off on runtime. I am not familiar with multiprocessing in Python, so I have been reading about different methodologies (Queue vs Pooling, etc.). I am curious which method would run best for this workflow, and how would I implement this? Ideally, a multiprocessed-translated version of this code, or a guide to get there would be great.

Thank you.

EDIT: in case I wasn’t clear, I want to run all 5 queries concurrently. Something that may be an issue is appending each DataFrame to the list concurrently, so if I need to I am willing to forgo that.

import pandas as pd
import psycopg2
import time
import os

host = os.environ["DBHOST"]
user = os.environ["DBUSER"]
pass = os.environ["DBPWD"]

db_conn = psycopg2.connect("host='{}' port={} dbname='{}' user={} password={}".format(host, 
                                                                                           port#, 
                                                                                           "db_name", 
                                                                                           user, 
                                                                                           pass))
query_load = [("SELECT column_name_1, COUNT(*) "
            "FROM schema.table "
            "GROUP BY column_name_1 "
            "ORDER BY column_name_1 ASC"),

             ("SELECT column_name_2, COUNT(*) "
            "FROM schema.table "
            "GROUP BY column_name_2 "
            "ORDER BY column_name_2 ASC"),

             ("SELECT column_name_3, COUNT(*) "
            "FROM schema.table "
            "GROUP BY column_name_3 "
            "ORDER BY column_name_3 ASC"),

             ("SELECT column_name_4, COUNT(*) "
            "FROM schema.table "
            "GROUP BY column_name_4 "
            "ORDER BY column_name_4 ASC"),

            ("SELECT column_name_5, COUNT(*) "
            "FROM schema.table "
            "GROUP BY column_name_5 "
            "ORDER BY column_name_5 ASC")]
start_time = time.time()
data_load = []
for queries in query_load:
    data_load.append(pd.read_sql(queries, db_conn))
elapsed_time = time.time() - start_time
print ("Job finished in {} seconds".format(elapsed_time))

How do you know that starting SQL queries concurrently will mean that your DB will run them concurrently? — quamrana
– quamrana, Commented May 31, 2019 at 19:35
@quamrana I don't know for certain, but I don't have any reason to think it wouldn't.. do I? — Tom Hood
– Tom Hood, Commented May 31, 2019 at 19:40
If you can post some code, I’ll upvote and make as correct answer (assuming it works) haha. I don’t really know where to start with using Multiprocessing or Pooling. The online documentation didn’t quite click with me. — Tom Hood
– Tom Hood, Commented May 31, 2019 at 19:55
@TomHood, multiprocessing is more resource expensive, use asyncio instead — RomanPerekhrest
– RomanPerekhrest, Commented May 31, 2019 at 20:01

quamrana · Accepted Answer · 2019-05-31 20:43:29Z

3

Since you already have a collection of queries, we can organise a function to take one at a time, but by using Pool.map, they can run concurrently:

from multiprocessing import Pool
import pandas as pd
import time

# define query_load
# define db_conn

def read_sql(query):
    return pd.read_sql(query, db_conn)

if __name__ == '__main__':
    start_time = time.time()
    with Pool(5) as p:
        data_load = p.map(read_sql, query_load)
    elapsed_time = time.time() - start_time
    print ("Job finished in {} seconds".format(elapsed_time))
    # carry on re-processing data_load

Now, I'm assuming that db_conn will allow concurrent requests.

Also note that p.map does the organising of taking results and loading them into a list for you.

answered May 31, 2019 at 20:43

quamrana

39.5k13 gold badges57 silver badges77 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

Tom Hood Over a year ago

Looks very promising! I will try this on Monday!

Tom Hood Over a year ago

This works perfectly. I'm getting an estimated average acceleration factor of ~4, which makes sense on a 4-core cpu machine. Thank you!

Tom Hood Over a year ago

One final comment to anyone using this solution. I found with my initial attempts that I could not implement this in Jupyter Notebooks. The work around for this ( in my case) was to save my script on my desktop, then run the .py using cmd.

Zedi10 Over a year ago

I am trying to use the same principle to read 5 different queries in parallel and storing them in the 5 dataframes for data processing. The read is parallelized by this function, however the p.map closure i.e. the conversion of df to list is taking time. Any suggestions?

Arpit Saxena · Accepted Answer · 2021-07-12 13:33:39Z

1

(This is better as a comment to the accepted answer, but I don't have enough reputation for that)

From the psycopg2 documentation: https://www.psycopg.org/docs/usage.html#thread-safety

libpq connections shouldn’t be used by a forked processes, so when using a module such as multiprocessing or a forking web deploy method such as FastCGI make sure to create the connections after the fork.

So passing connection object to a pool is not safe, and instead a new connection should be created by each process.

answered Jul 12, 2021 at 13:33

Arpit Saxena

673 silver badges7 bronze badges

Collectives™ on Stack Overflow

Python using multiprocessing on PostgreSQL queries reduce runtime

2 Answers 2

4 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

4 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related