python - 複数のジオプロセスでマルチプロセッシングを実装する方法は?

Question

私はPythonに比較的慣れていないので、マルチプロセッシングを試してみようと思いました。IDLE または ArcMap ツールボックス スクリプトとして適切に実行されるスクリプトがあります。これらのフォーラムと docs.python を熟読した後、作業スクリプトをマルチプロセッシング スクリプトに組み込んでみました。ただし、このフォーラムの同様の作業例はありますが、私が望むようにデータ処理に対処するものはありません。実現可能であることを願っています。

基本的に、スクリプトは標高ラスター (ERDAS IMG 形式) のリストを移動し、しきい値を下回るセルを抽出し、最後にそれらを結合します。現在、コマンド プロンプトでスクリプトを実行しています。他のすべてのウィンドウが新しいウィンドウを開くか、クラッシュしようとしてクラッシュします。スクリプトは、ワーカーが完全に終了するのを待つ前に最終的なマージに進むように見えることを除けば、正常に動作するように見えます。

私はいくつかの例を見てきましたが、ワーカー関数に 2 つ以上のプロセスがあるように見えるものはほとんどありません。いずれもアークピー ジオプロセスではありません。したがって、私の質問は基本的に 1) pool.map や pool.apply など、pool.apply_async 以外のものを使用する必要がありますか? 2) 最終的なポリゴンのパスを resultList に適切に返していますか?

どんな批判も大歓迎です。前もって感謝します。

# Import modules
import arcpy, os, math
from arcpy import env
from arcpy.sa import *
import multiprocessing
import time

# Check out licenses
arcpy.CheckOutExtension("spatial")

# Define functions
def worker_bee(inputRaster, scratch, addNum):   
    (path, lName) = os.path.split(inputRaster)
    (sName, ext) = os.path.splitext(lName)
    nameParts = sName.split("_")
    nameNumber = nameParts[-1]

    # Create scratch subfolder if not exists
    subFolder = scratch + "\\" + nameNumber + "_output" 
    if not os.path.exists(subFolder):os.makedirs(subFolder)
    # Set workspace to subfolder
    arcpy.env.workspace = subFolder
    arcpy.env.overwriteOutput=True
    arcpy.env.extent = "MAXOF"

    # Local Variables
    Expression = "Shape_Area >= 100"
    poly1 = subFolder + "\\poly1.shp"
    poly2 = subFolder + "\\poly2.shp"
    poly3 = subFolder + "\\poly3.shp"
    poly4 = subFolder + "\\poly4.shp"
    poly5 = subFolder + "\\poly5.shp"
    poly6 = subFolder + "\\poly6.shp"
    poly7 = subFolder + "\\poly7.shp"
    outName = scratch + "\\ABL_" + nameNumber + ".shp"

    #### Perform calculations ###
    # Map Algebra (replace -9999 with 9999)
    inRasterCon = Con(inputRaster, 9999, inputRaster, "Value = -9999")
    # Filter DEM to smooth out low outliers
    filterOut = Filter(inRasterCon, "LOW", "DATA")
    # Determine raster MINIMUM value and calculate threshold
    filterMinResult = arcpy.GetRasterProperties_management(filterOut, "MINIMUM")
    filterMin = filterMinResult.getOutput(0)
    threshold = (float(filterMin) + float(addNum))
    # Map Algebra (values under threshold)
    outCon = Con(filterOut <= threshold, 1, "")
    arcpy.RasterToPolygon_conversion(outCon, poly1, "SIMPLIFY", "Value")
    # Dissolve parts
    arcpy.Dissolve_management(poly1, poly2, "", "", "SINGLE_PART", "DISSOLVE_LINES")    
    # Select parts larger than 100 sq m
    arcpy.Select_analysis(poly2, poly3, Expression)
    # Process: Eliminate Polygon Part
    arcpy.EliminatePolygonPart_management(poly4, poly5, "PERCENT", "0 SquareMeters", "10", "CONTAINED_ONLY")
    # Select parts larget than 100 sq m
    arcpy.Select_analysis(poly5, poly6, Expression)
    # Simplify Polygon
    arcpy.SimplifyPolygon_cartography(poly6, poly7, "BEND_SIMPLIFY", "3 Meters", "3000 SquareMeters", "RESOLVE_ERRORS", "KEEP_COLLAPSED_POINTS")
    # Smooth Polygon
    outShape = arcpy.SmoothPolygon_cartography(poly7, outName, "PAEK", "3 Meters", "FIXED_ENDPOINT", "FLAG_ERRORS").getOutput(0)
    ### Calculations complete ###

    # Delete scratch subfolder
    arcpy.Delete_management(subFolder)

    print("Completed " + outShape + "...")
    return outShape

resultList = []
def log_result(result):
    resultList.append(result)

if __name__ == "__main__":
    arcpy.env.overwriteOutput=True
    # Read in parameters
    inFolder = raw_input("Input Folder: ")#arcpy.GetParameterAsText(0)
    addElev = raw_input("Number of elevation units to add to minimum: ")

    # Create scratch folder workspace
    scratchFolder = inFolder + "\\scratch" 
    if not os.path.exists(scratchFolder):os.makedirs(scratchFolder)

    # Local variables
    dec_num = str(float(addElev) - int(float(addElev)))[1:]
    outNameNum = dec_num.replace(".", "")
    outMerge = inFolder + "\\ABL_" + outNameNum + ".shp"

    # Print core usage
    cores = multiprocessing.cpu_count()
    print("Using " + str(cores) + " cores...")

    #Start timing
    start = time.clock()

    # List input tiles
    arcpy.env.workspace = inFolder
    inTiles = arcpy.ListRasters("*", "IMG")
    tileList = []
    for tile in inTiles:
        tileList.append(inFolder + "\\" + tile)

    # Create a Pool of subprocesses
    pool = multiprocessing.Pool(cores)

    print("Adding jobs to multiprocessing pool...")
    for tile in tileList:
        # Add the job to the multiprocessing pool asynchronously
        pool.apply_async(worker_bee, (tile, scratchFolder, addElev), callback = log_result)
    pool.close()
    pool.join()

    # Merge the temporary outputs
    print("Merging temporary outputs into shapefile " + outMerge + "...")
    arcpy.Merge_management(resultList, outMerge)

    # Clean up temporary data
    print("Deleting temporary data ...")
    for result in results:
        try:
            arcpy.Delete_management(result)
        except:
            pass

    # Stop timing and report duration
    end = time.clock()
    duration = end - start
    hours, remainder = divmod(duration, 3600)
    minutes, seconds = divmod(remainder, 60)
    print("Completed in " + hours + "hrs " + minutes + "min " + seconds + "sec")