Multiprocessing with geopandas

Question

I am learning how to implement the multiprocessing with spatial data using the module multiprocessing.

I am trying to implement a spatial intersection between a polygon file and a grid file. I am reading the data with geopandas only reading input files took 3 minutes: the grid's .shp file is 806 MB.

I am trying to implement this code(it is working ):

    data = []
    for index, orig in grid.iterrows():  # iterate over row on a dataframe
        for index2, ref in polygons.iterrows():
            if ref['geometry'].intersects(orig['geometry']): # if the feature intersect each other
             grid_id = orig['Id']
             gid_id = ref['gid_id']
             gid_pk = ref['gid_pk']

    data.append({'geometry':ref['geometry'].intersection(orig['geometry']),'grid_id':grid_id,
                 'nuts3_id': gid_id, 'nuts3_pk': gid_pk})
    df = gpd.GeoDataFrame(data,columns=['geometry','grid_id', 'nuts3_id', 'nuts3_pk'])
    df.to_file('intersection.shp')

Multiprocessing code

def chunks(l, n):
    for i in range(0, len(l), n):
        # yield l.loc[i:i+n ,:]
        f = l.loc[i:i+n -1 ,:]
        return f

def intersect_geom(z, polygons, d):
    for index2, ref in polygons.iterrows():

        if ref['geometry'].intersects(z['geometry']): # if the feature intersect each other

            grid_id = z['Id']
            gid_id = ref['gid_id']
            gid_pk = ref['gid_pk']
            inter_geom = ({'geometry':ref['geometry'].intersection(z['geometry']),'grid_id':grid_id,
             'nuts3_id': gid_id, 'nuts3_pk': gid_pk})


def main():
    data = []

    man = mp.Manager()
    d = man.dict()
    split = chunks(grid, len(grid)//10)
    for pos, z in split.iterrows():
        p = mp.Process(target=intersect_geom, args=(z, polygons, d))
        p.start()
        data.append(p)
        #
        # # wait that all chunks are finished
        [j.join() for j in data]




if __name__ == '__main__':

    main()

I got this error back

 Traceback (most recent call last):
  File "test_2.py", line 70, in <module>
    main()
  File "test_2.py", line 57, in main
    p.start()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 105, in start
    self._popen = self._Popen(self)
  File "/usr/lib/python3.6/multiprocessing/context.py", line 223, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "/usr/lib/python3.6/multiprocessing/context.py", line 277, in _Popen
    return Popen(process_obj)
  File "/usr/lib/python3.6/multiprocessing/popen_fork.py", line 19, in __init__
    self._launch(process_obj)
  File "/usr/lib/python3.6/multiprocessing/popen_fork.py", line 65, in _launch
    parent_r, child_w = os.pipe()
OSError: [Errno 24] Too many open files
Error in sys.excepthook:
Traceback (most recent call last):
  File "/usr/lib/python3/dist-packages/apport_python_hook.py", line 63, in apport_excepthook
    from apport.fileutils import likely_packaged, get_recent_crashes
  File "/usr/lib/python3/dist-packages/apport/__init__.py", line 5, in <module>
    from apport.report import Report
  File "/usr/lib/python3/dist-packages/apport/report.py", line 30, in <module>
    import apport.fileutils
  File "/usr/lib/python3/dist-packages/apport/fileutils.py", line 23, in <module>
    from apport.packaging_impl import impl as packaging
  File "/usr/lib/python3/dist-packages/apport/packaging_impl.py", line 24, in <module>
    import apt
  File "/usr/lib/python3/dist-packages/apt/__init__.py", line 35, in <module>
    apt_pkg.init_system()
apt_pkg.Error: E:Error reading the Tuple table

Original exception was:
Traceback (most recent call last):
  File "test_2.py", line 70, in <module>
    main()
  File "test_2.py", line 57, in main
    p.start()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 105, in start
    self._popen = self._Popen(self)
  File "/usr/lib/python3.6/multiprocessing/context.py", line 223, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "/usr/lib/python3.6/multiprocessing/context.py", line 277, in _Popen
    return Popen(process_obj)
  File "/usr/lib/python3.6/multiprocessing/popen_fork.py", line 19, in __init__
    self._launch(process_obj)
  File "/usr/lib/python3.6/multiprocessing/popen_fork.py", line 65, in _launch
    parent_r, child_w = os.pipe()
OSError: [Errno 24] Too many open files

How can I fix this error and make the code run on multiprocessing?

Answer 1

Two things that need to be addressed first:

1. the title of your question is far too broad and misleading with respect to your actual problem.
2. if you have spatial operations such as spatial joins, or in your case intersections, multiprocessing can lead to a loss of successful spatial joins.

Since your question asks about multiprocessing and geopandas in general, I'll take the liberty of answering the question in general, step by step:

Data Preparation. When doing spatial operations between two geopandas dataframes, you can't just split the two and do multiprocessing, because then you'd just be trying to spatially connect certain chunks together. One can speed this up by creating chunks according to specific geographic subregions, hopefully you have such a feature. First create a dictionary which splits your data by such a feature. So a key is the label of this feature and the value contains the dataframe which has for the certain feature only the key value.

import pandas as pd

#create unique list of feature which is in both geodataframes
uniqueFeatureInLeftDF = gdfLeft.FeatureInLeftDF.unique()

#create a data frame dictionary to store your data frames
gdfLeftDict = {elem : pd.DataFrame() for elem in uniqueFeatureInLeftDF }

for key in gdfLeftDict.keys():
    gdfLeftDict[key] = gdfLeft[:][gdfLeft.FeatureInLeftDF== key]

Do the above code again for the gdf you want to be joined, however in this case the value do not have to vary, since they will be filtered later on:

gdfRigthDict = {elem : pd.DataFrame() for elem in uniqueFeatureInLeftDF }
for key in gdfRigthDict.keys():
    gdfRigthDict[key] = gdfRigth

Now the data can be filtered by creating a convexhull for each unique value of the feature which splitted the left dataframe. Then this convexhull is used to reduce the right dataframe to rows which intersect the corresponding convexhull. (Concavhull would be better but at the moment geopandas has not such a built in function).

import geopandas as gpd
from geopandas.tools import sjoin

for k, v in gdfLeftDict.items():

    #create convexhull for the left geodfs
    geometryLeft = v.geometry
    gdfConvexhull = gpd.GeoDataFrame(geometry=geometryLeft)
    gdfConvexhull = gdfConvexhull.set_crs('epsg:XXXX')
    convexhull = gdfConvexhull.unary_union.convex_hull
    df_convexhull = pd.DataFrame({'FeatureInLeftDF':k, 'geometry':[convexhull]})
    gdf_convexhull = gpd.GeoDataFrame(df_convexhull, geometry='geometry')
    gdf_convexhull = gdf_convexhull.set_crs('epsg:XXXX')
    gdf_convexhull.sindex

    #select the rigth dataframe for the current featrue/chunk
    gdfRigthBeforeFiltering = gdfRigthDict[k]

    #spatial index the data
    gdfRigthBeforeFiltering.sindex

    #filter the data
    gdfRigthAfterFiltering = sjoin(
            gdfRigthBeforeFiltering,
            gdf_convexhull,
            how="inner", #not left or rigth!
            op="intersects",
            )
    gdfRigthAfterFiltering.drop("index_right", axis=1, inplace=True)
    
    #update the rigth dictionary with the filtered data
    gdfRigthDict[k] = gdfRigthAfterFiltering

In general, in order to speed up a join, filter your dataframes first! So if date is also a criteria, you could also filter your dataframes by date before applying spatial operations.

Multiprocessing function Now you need to define the function that you want to multiprocess later. Let's take a normal spatial join as an example:

def multiprocessSpatialJoin(gdfLeft, gdfRigth): 
    
    gdf_LeftJoin = sjoin(
            gdfLeft,
            gdfRigth,
            how="left",
            op="intersects",
            )
    gdf_LeftJoin.drop("index_right", axis=1, inplace=True)

    return gdf_LeftJoin

If you are working on Windows with Jupyter notebooks, make sure you save this function in a separate script and import it for the multiprocessing part (in this example, the function is saved in a pythonScripts folder, in a file called multiprocessFunctions.py)!

Multiprocessing Now it's time to speed up the spatial join, which otherwise would have taken forever:

import multiprocessing as mp
from pythonScripts.multiprocessFunctions import multiprocessSpatialJoin

if __name__ == '__main__': 

    #5=number of cores used 
    p = mp.Pool(5)
    #starmap allows to insert two variables, the order and length of the two lists must be the same
    results = p.starmap(multiprocessSpatialJoin, zip(list(gdf_pointsDict.values()), list(gdf_linesDict.values())))
    #stop running
    p.close()
    #join the results
    p.join()

    #store the joined results if necessary can be concatet
    gdf_spatial_joins_list = [x for x in results]