并行迭代pandas df

Question

所以，我想并行迭代一个 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 

    """

Answer 1

不要启动 15 个进程，而是使用线程并使用参数调用线程函数。threading.Thread(target=func, args=(i,))其中 i 是您的号码，func是包装整个代码的函数。然后遍历它。您不需要在 15 个项目处并行化迭代。