Python: Multiprocessing error on pandas data frame: Clients have non-trivial state that is local and unpickleable

user_4899518 0
Viewed 735

I have a dataframe that I am dividing into multiple dataframes using groupby. Now I want to process each of these dataframes for which I have written a function process_s2id in parallel. I have the whole code in a class which I am executing using a main function in another file. But I am getting the following error:

 "Clients have non-trivial state that is local and unpickleable.",
_pickle.PicklingError: Pickling client objects is explicitly not supported.
Clients have non-trivial state that is local and unpickleable.

Following is the code (we execute the main() function in this class):

import logging
import pandas as pd
from functools import partial
from multiprocessing import Pool, cpu_count

class TestClass:
    
    def __init__(self):
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger()
        
    def process_s2id(self, df, col, new_col):
        dim2 = ['s2id', 'date', 'hours']
        df_hour = df.groupby(dim2)[[col, 'orders']].sum().reset_index()
        df_hour[new_col] = df_hour[col] / df_hour['orders']
        df_hour = df_hour[dim2 + [new_col]]
        return df_hour
    
    def run_parallel(self, df):
        series = [frame for keys, frame in df.groupby('s2id')]

        p = Pool(cpu_count())
        prod_x = partial(
            self.process_s2id,
            col ="total_supply",
            new_col = "supply"
        )
        s2id_supply_list = p.map(prod_x, series)
        p.close()
        p.join()

        s2id_supply = pd.concat(s2id_supply_list, axis=0)
        return ms2id_bsl
    
    def main(self):
        data = pd.read_csv("data/interim/fs.csv")
        out = self.run_parallel(data)
        return out

I tried running this code in Spyder and it works fine. But when I am executing it from another file. I am getting an error. Following are the execution file code and error:

import TestClass

def main():
    tc = TestClass()
    data = tc.main()

if __name__ == '__main__':
    main()

When I looked into the error traceback, I found that the error is occuring on the line s2id_supply_list = p.map(prod_x, series) where the function is starting to go parallel. I also tried running this in series and it worked. Also, I noticed that this particular error is coming from client.py from Google cloud package. There is a certain code in which I am uploading the data to Google cloud but that should be invariant to this code. I tried searching hard for this error but all the results are linked to Google cloud package related issues and not the multiprocessing package.

Can anyone help me out in understanding this error and how can I fix it?

Other information: I have the following versions of packages:

python==3.7.7
pandas==1.0.5
google-cloud-storage==1.20.0
google-cloud-core==1.0.3

I am running this on macbook pro.

pandas pickle google-cloud-platform multiprocessing python
1 Answers

I figured out. When we are using Pool over a function to run it parallel, it expects the first argument to be the iterator. In other words, the function will run parallelly over different values of the first argument. When we have a non-static function in a class, we have the first argument as self or the class itself. But the stupid Pool function doesn't know how to iterate with the self because it's the wrong argument. The right argument is the second one.

We can solve this by either:

  1. Taking the function out of the class and kicking the self out of the arguments.
  2. Adding @staticmethod on top of the function and kicking the self out of the arguments.

I hope this helps someone who is struggling with a similar problem.

user_48995180
Related