pool.apply_asyn c имеет только один процесс

Question

Я написал код для очистки данных. Операция c - это «single_worker», присваивающая значения от «a» до «l», если первый столбец больше, чем точки среза. Поскольку весь набор данных довольно большой, я разделил его на 8 подмножеств для этого метода многопроцессорной обработки. Тем не менее, когда я запускаю этот код в cmd, на самом деле он имеет только один процесс, работающий с подмножеством 1, и когда этот процесс завершается, дальнейший процесс для подмножества 2.

Моя python версия 3.7 со всеми анакондами пакеты. Компьютерная система win10 с 12 ядрами.

Вот функция:

    def single_worker(merge, beme, me,a):
        c = 0
        year_1 = 0
        month_1 = 0
        length = len(merge)
        for i in trange(length):
        c = c + 1
        year = merge['year'][i]
        month = merge['month'][i]
        if (year != year_1) | (month != month_1):
            key_beme = beme[beme.year == year].index.tolist()
            k=key_beme[0]
            beme6 = beme['6'][k]
            beme14 = beme['14'][k]

            key_me = me[(me.year==year)&(me.month==month)].index.tolist()
            b=key_me[0]
            me10 = me['10'][b]

            year_1 = year
            month_1 = month


        if ~(merge['me'][i] > 0):
            merge['bs_new'][i] = np.nan
            continue
        if merge['me'][i] <= me10:
            merge['bs_new'][i] = 's'
        else:
            merge['bs_new'][i] = 'b'

        if ~(merge['bm'][i] > 0):
            merge['hl_new'][i] = np.nan
            continue
        if merge['bm'][i] <= beme6:
            merge['hl_new'][i] = 'l'
        elif merge['bm'][i] >= beme14:
            merge['hl_new'][i] = 'h'
        else:
            merge['hl_new'][i] = 'm'
        name = str(a)+".csv"
        merge.to_csv(name)

Вот код:

    import pandas as pd
    import numpy as np


    from tqdm import trange                  
    from multiprocessing import cpu_count    
    from multiprocessing import Pool
    from single_worker import single_worker

    merge = pd.read_csv('merge.csv')
    beme = pd.read_csv('beme.csv')
    me = pd.read_csv('me.csv')
    Len_imgs = len(merge)  
    num_cores = cpu_count()  


    if num_cores >= 8:  
    num_cores = 8
    subset1 = merge[:Len_imgs // 8]
    subset2 = merge[Len_imgs // 8: Len_imgs // 4]
    subset3 = merge[Len_imgs // 4: (Len_imgs * 3) // 8]
    subset4 = merge[(Len_imgs * 3) // 8: Len_imgs // 2]
    subset5 = merge[Len_imgs // 2: (Len_imgs * 5) // 8]
    subset6 = merge[(Len_imgs * 5) // 8: (Len_imgs * 6) // 8]
    subset7 = merge[(Len_imgs * 6) // 8: (Len_imgs * 7) // 8]
    subset8 = merge[(Len_imgs * 7) // 8:]

    List_subsets = [subset1, subset2, subset3, subset4,
                    subset5, subset6, subset7, subset8]
    print("Finish separating subsets")
    p = Pool(num_cores)

    k=0

    for i in range(num_cores):
        k=k+1
        p.apply_async(single_worker, (List_subsets[i], beme, me,k,))

        print(k)
    p.close()
    p.join()

Кстати, когда это было работает, я проверил диспетчер задач. Загрузка ЦП постоянно ниже 25%. Я не знаю, если это проблема с моим кодом. Взгляни, пожалуйста. Спасибо за ваше время.

Answer 1

Я пытался не использовать trange в определенной мной функции, и она работает. Это кажется странным, и я не знаю причину.