python之并行迭代 pandas df

所以，我想并行遍历 pandas df，所以假设我有 15 行，然后我想并行遍历它，而不是逐行遍历。

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' } 
    
]) 

因此，我遍历 df 并制作命令行，然后将输出存储在 df 中并进行数据过滤，最后将其存储到 influxdb 中。问题是我在迭代它时一个一个地做。我想并行遍历所有行。

截至目前，我已经制作了 20 个脚本并使用多处理并行处理所有脚本。当我必须在所有 20 个脚本中进行更改时，这很痛苦。我的脚本如下所示:-

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  
    """ 

截至目前，如果我在 df 中运行 15 行并进行如下所示的并行处理，我将拥有 15 个脚本:-

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) 

问题:- 我如何遍历 df 并使所有 15 行并行运行并在 for 循环内执行所有操作。

请您参考如下方法:

我将从 Reddit 上复制并粘贴我的答案(以防有人遇到类似情况):

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') 

您可能需要在 apply 方法中使用轴参数。