2

So, I want to iterate over a pandas df in parallel so suppose i am having 15 rows then i want to iterate over it parallel and not one by one.

df:-

df = pd.DataFrame.from_records([ {'domain':'dnd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'hrpd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'blhp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'rbswp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'foxbp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'rbsxbp','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'dnd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' }, {'domain':'hrpd','duration':'90','media_file':'testfont.wav','user':'tester_food','channel':'confctl-2' } ])

enter image description here

So, I am iterating over the df and making command line and then storing the output in a df and doing data filtering and then finally storing it into influxdb. The problem is i am doing it one by one as i am iterating over it. what i want to iterate over all the rows in parallel.

As of now i have made 20 scripts and using multiprocessing to go over all the scripts in parallel. Its a pain when i have to do a change as i have to do it in all 20 scripts. My script looks like below :-

for index, row in dff.iterrows(): domain = row['domain'] duration = str(row['duration']) media_file = row['media_file'] user = row['user'] channel = row['channel'] cmda = './vaa -s https://' + domain + '.www.vivox.com/api2/ -d ' + duration + ' -f ' + media_file + ' -u .' + user + '. -c sip:confctl-2@' + domain + '.localhost.com -ati 0ps-host -atk 0ps- test' rows = [shlex.split(line) for line in os.popen( cmda).read().splitlines() if line.strip()] df = pd.DataFrame(rows) """ Bunch of data filteration and pushing it into influx """

As of now i am having 15 script if i am hvaing 15 rows in df and doing parallel processing like below :-

import os import time from multiprocessing import Process os.chdir('/Users/akumar/vivox-sdk-4.9.0002.30719.ebb523a9') def run_program(cmd): # Function that processes will run os.system(cmd) # Creating command to run commands = ['python testv.py'] commands.extend(['python testv{}.py'.format(i) for i in range(1, 15)]) # Amount of times your programs will run runs = 1 for run in range(runs): # Initiating Processes with desired arguments running_programs = [] for command in commands: running_programs.append(Process(target=run_program, args=(command,))) running_programs[-1].daemon = True # Start our processes simultaneously for program in running_programs: program.start() # Wait untill all programs are done while any(program.is_alive() for program in running_programs): time.sleep(1)

Question:- How Can i iterate over the df and make all the 15 rows to run in parallel and do all the stuff inside the for loop.

Improve this question

2 Answers 2

Reset to default
8

I'm gonna copy and paste my answer from Reddit on here (in case anyone stumbles upon it with a similar situation):

import dask.dataframe as ddf def your_function(row): domain = row['domain'] duration = str(row['duration']) media_file = row['media_file'] user = row['user'] channel = row['channel'] cmda = './vaa -s https://' + domain + '.www.vivox.com/api2/ -d ' + duration + ' -f ' + media_file + ' -u .' + user + '. -c sip:confctl-2@' + domain + '.localhost.com -ati 0ps-host -atk 0ps- test' rows = [shlex.split(line) for line in os.popen( cmda).read().splitlines() if line.strip()] df_dask = ddf.from_pandas(df, npartitions=4) # where the number of partitions is the number of cores you want to use df_dask['output'] = df_dask.apply(lambda x: your_function(x), meta=('str')).compute(scheduler='multiprocessing')

You might have to play around with the axis parameter in the apply method.

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

2 Comments

UPDATE: Another good way to do multiprocessing for Pandas dataframe's is the modin project: github.com/modin-project/modin
In your last row, lambda x: your_function(x) can be simplified to just your_function
0

Instead of starting 15 processes, use threading and call the threaded function with an argument. threading.Thread(target=func, args=(i,)) where i is your number and func is function that wraps the whole code. Then iterate through it. You don't need to parallelize the iteration at 15 items.

Improve this answer

Comments