Python: Parallelize nested for loop

user_13465932 0
Viewed 80

I have nested for-loop in python to create a netCDF file. The for-loop takes a pandas dataframe with time, lat, lot, and parameters and replaces the information in the netCDF file by the parameters in the correct location and time. This is taking too long since the pandas dataframe has more than 80000 rows and the netCDF file has around 8000 time-steps. I've been looking to use either xargs or multiprocessing but in the first case the uses files as inputs and in the second case, it produces as many outputs as processes I use. I have no experience in parallel processing so probably my affirmations are totally wrong. This is the code that I am using:

with Dataset(os.path.join('Downloads', inv, 'observations.nc'), 'w') as dset:
    
    dset.createDimension('time_components', 6)
    
    groups = ['obs', 'mix_apri', 'mix_apos', 'mix_background']
    
    for group in groups:
        
        dset.createGroup(group)
        dset[group].createDimension('nt', 8760)
        dset[group].createDimension('nlat', 80)
        dset[group].createDimension('nlon', 100)
        
        times_start = dset[group].createVariable('times_start', 'i4', ('nt', 'time_components'))
        times_end = dset[group].createVariable('times_end', 'i4', ('nt', 'time_components'))
        lats = dset[group].createVariable('lats', 'f4', ('nlat'))
        lons = dset[group].createVariable('lons', 'f4', ('nlon'))
        
        times_start[:,:] = list(emis_apri['biosphere']['times_start'])
        times_end[:,:] = list(emis_apri['biosphere']['times_end'])
        lats[:] = list(emis_apri['biosphere']['lats'])
        lons[:] = list(emis_apri['biosphere']['lons'])
        
    conc_obs = dset['obs'].createVariable('conc', 'f8', ('nt', 'nlat', 'nlon'))
    conc_mix_apri = dset['mix_apri'].createVariable('conc', 'f8', ('nt', 'nlat', 'nlon'))
    conc_mix_apos = dset['mix_apos'].createVariable('conc', 'f8', ('nt', 'nlat', 'nlon'))
    conc_mix_background = dset['mix_background'].createVariable('conc', 'f8', ('nt', 'nlat', 'nlon'))
    
    for i in range(8760):
        conc_obs[i,:,:] = emis_apri['biosphere']['emis'][i][:,:]*0
        
    conc_mix_apri[:,:,:] = list(conc_obs)
    conc_mix_apos[:,:,:] = list(conc_obs)
    conc_mix_background[:,:,:] = list(conc_obs)
    
    db = obsdb(os.path.join('Downloads', inv, 'observations.apos.tar.gz'))
    nsites = db.sites.shape[0]
    for isite, site in enumerate(db.sites.itertuples()):
        dbs = db.observations.loc[db.observations.site == site.Index]
        lat = where((array(emis_apri['biosphere']['lats']) >= list(dbs.lat)[0]-0.25) & (array(emis_apri['biosphere']['lats']) <= list(dbs.lat)[0]+0.25))[0][0]
        lon = where((array(emis_apri['biosphere']['lons']) >= list(dbs.lon)[0]-0.25) & (array(emis_apri['biosphere']['lons']) <= list(dbs.lon)[0]+0.25))[0][0]
        for i in range(len(list(dbs.time))):
            for j in range(len(times_start)):
                if datetime(*times_start[j,:].data) >= Timestamp.to_pydatetime(list(dbs.time)[i]) and datetime(*times_end[j,:].data) >= Timestamp.to_pydatetime(list(dbs.time)[i]):
                    conc_obs[i,lat,lon] = list(dbs.obs)[i]
                    conc_mix_apri[i,lat,lon] = list(dbs.mix_apri)[i]
                    conc_mix_apos[i,lat,lon] = list(dbs.mix_apos)[i]
                    conc_mix_background[i,lat,lon] = list(dbs.mix_background)[i]

From for isite, site in enumerate(db.sites.itertuples()): is the part of the code that I need to parallelize. I really appreciate any insights about this.

pandas python-multiprocessing netcdf xargs python
1 Answers

Consider the following as pseudocode as I cannot run any test without any samples etc. I have usually parallelized my code with mpi4py and in your case, you could do in the beginning:

from mpi4py import MPI
comm = MPI.COMM_WORLD
size = comm.Get_size(); # let your program know how many processors you are using
rank = comm.Get_rank() # let the running program know, which processor it is

Now, in the beginning of the code, let one of the processes be the so called master task, which can do all the basic/important stuff that cannot be done simultaneously by all the tasks. For example, opening/initializing some file for the output. So, in your code, for those parts, you can use:

if rank==0:
    # do some important stuff
else:
    # do something not important (for example a = 5)
comm.barrier() # this is important to synchronize the  processes

Now, to parallelize your code, you can do the loop over distributed db.sites i.e. you divide the db.sites.itertuples() over the number of processor you are going to use:

allsites = db.sites.itertuples() # all the processor have to know all the sites
sites = allsites[rank::size] # each starts from it's current rank and jumps with the size

for isite, site in enumerate(sites):
    dbs = db.observations.loc[db.observations.site == site.Index]
    lat = where((array(emis_apri['biosphere']['lats']) >= list(dbs.lat)[0]-0.25) & (array(emis_apri['biosphere']['lats']) <= list(dbs.lat)[0]+0.25))[0][0]
    lon = where((array(emis_apri['biosphere']['lons']) >= list(dbs.lon)[0]-0.25) & (array(emis_apri['biosphere']['lons']) <= list(dbs.lon)[0]+0.25))[0][0]
    for i in range(len(list(dbs.time))):
        for j in range(len(times_start)):
            if datetime(*times_start[j,:].data) >= Timestamp.to_pydatetime(list(dbs.time)[i]) and datetime(*times_end[j,:].data) >= Timestamp.to_pydatetime(list(dbs.time)[i]):
                conc_obs[i,lat,lon] = list(dbs.obs)[i]
                conc_mix_apri[i,lat,lon] = list(dbs.mix_apri)[i]
                conc_mix_apos[i,lat,lon] = list(dbs.mix_apos)[i]
                conc_mix_background[i,lat,lon] = list(dbs.mix_background)[i]

comm.barrier() # do not forget to synchronize

Nevertheless, in this case the "isite" has now value based on the size of the list, you are giving in. So instead of being 0...len(allsites), it is 0...len(allsites)/size. If the "isite" is important to have value from 0 to len(allsites), you somehow have to recalculate. Perhaps isite_global = isite*size+rank to get the actual number that the processor is doing.

So, in the end how to run the code, I usually do:

mpiexec -np 10 ipython script_name

at the terminal to run the code on 10 processors.

But, in any case, the hardest part is to parallelize the I/O operations without specific support from the library. I am not sure that netCDF4 supports parallel I/O meaning if your processors with ranks 0...X open the file simultaneously for X processors, write something to the specific location in the file and close the file, that afterwards the data from all the processors is written there.

Therefore, the safest idea is to somehow let one of the processors (master) be responsible for the output and exchange/collect data that needs to be written from all the subprocessors before writing.

Hope this helps, good luck with the code!

user_19133670
Related