如何并行裁剪栅格？

Question

我正在处理一个多处理示例 (An introduction to parallel programming)。我修改了 Pool Class 示例以满足我的特定需求 - 以平行的研究区域多边形裁剪一堆栅格。从好的方面来说，脚本完成并打印“处理完成”。不利的一面是，没有产生输出。我怀疑我在“pool.apply_async”函数中有一些程序错误。为什么这个脚本没有结果？

import arcpy, os
import multiprocessing as mp

arcpy.env.workspace = r'F:\temp\inws'
outws_utm11 = r'F:\temp\outws'
clipper_utm11 = r'F:\temp\some_polygon.shp'

rasters = arcpy.ListRasters()

pool = mp.Pool(processes=4)

def clip_raster(clipper, outws, raster):
        arcpy.Clip_management(raster, "#", os.path.join(outws, raster), clipper, nodata_value = 0, clipping_geometry = "ClippingGeometry")

[pool.apply_async(clip_raster, args=(clipper_utm11,outws_utm11, ras)) for ras in rasters]

print "Processing complete."

Answer 1

apply_async 函数在工作进程中启动您的函数，但在函数完成之前不会阻塞。您让主进程完成并退出，而不是等待工作人员完成。这会导致他们被杀死，这很可能在他们创建你的输出之前发生。

由于您只是对rasters 列表中的所有项目应用相同的函数，因此您应该考虑改用pool.map。它将接受函数名称和可迭代对象作为其参数，并在列表中的每个项目上调用该函数。所有这些函数调用都将发生在池中的一个工作进程中。不过，pool.map 函数的一个警告是，您传递给它的函数对象必须只接受一个参数：列表中的项目。我看到您的 clip_rasters 函数使用了几个其他参数，因此在下面的示例中，我使用 functools.partial 创建一个新版本的 clip_raster，始终包含前两个参数.这个绑定了clipper_utm11 和outws_utm11 的新函数现在可以与pool.map 一起使用。

import arcpy, os
import functools
import multiprocessing as mp

arcpy.env.workspace = r'F:\temp\inws'
outws_utm11 = r'F:\temp\outws'
clipper_utm11 = r'F:\temp\some_polygon.shp'

rasters = arcpy.ListRasters()

pool = mp.Pool(processes=4)

def clip_raster(clipper, outws, raster):
        arcpy.Clip_management(raster, "#", os.path.join(outws, raster), clipper, nodata_value = 0, clipping_geometry = "ClippingGeometry")

bound_clip_raster = functools.partial(clip_raster, clipper_utm11, outws_utm11)
results = pool.map(bound_clip_raster, rasters)

print "Processing complete."

此代码将为rasters 列表中的每个项目调用一次bound_clip_raster 函数，包括clipper_utm11 和outws_utm11。所有结果都将在名为results 的列表中提供，并且对pool.map 的调用处于阻塞状态，因此主进程将等到所有工作人员完成后才退出。

如果出于某种奇怪的原因，您打算使用apply_async，那么您需要在脚本末尾添加一些代码，以使用AsyncResult 对象的关联方法来阻止主进程，直到他们可以完成，例如wait()，或者通过调用ready() 在循环中轮询完成。但是对于这个用例，你真的应该使用pool.map。这就是它的用途。

Answer 2

我已经回答了一个可能对您也有用的问题。看这里：question

有一些好的做法，比如像你一样把所有东西都放在一个函数中，但也必须要放置一个main() 函数并用

if __name__ == '__main__':
    main()

另一件事是调用我在您的代码中插入的 pool.apply_assync 的一个函数。

还做了一些修改，所以你可以试试，我已经测试过，它对我有用：

import arcpy, os
from multiprocessing import Pool

arcpy.env.workspace = r'C:\Gis\lab_geo\2236622'
outws_utm11 = r'C:\Gis\lab_geo\2236622\outs'
clipper_utm11 = r'C:\Gis\arcpy_teste\teste.shp'

rasters = arcpy.ListRasters()

def clipRaster(clipper, outws, raster):
     arcpy.Clip_management(raster, "#", os.path.join(outws, raster), clipper, 0, "ClippingGeometry")

def clipRasterMulti(processList):
    pool = Pool(processes=4, maxtasksperchild=10)
    jobs= {}
    for item in processList:
        jobs[item[0]] = pool.apply_async(clipRaster, [x for x in item])
    for item,result in jobs.items():
        try:
            result = result.get()
        except Exception as e:
            print(e)
    pool.close()
    pool.join()

def main():
    processList = [(clipper_utm11,outws_utm11, ras) for ras in rasters]
    clipRasterMulti(processList)
    print "Processing complete."

if __name__ == '__main__':
    main()