ArcGIS Python批量化处理入门指南：从数据加载到自动化输出

1次阅读

没有评论

共计 2560 个字符，预计需要花费 7 分钟才能阅读完成。

在处理 GIS 数据时，手动操作不仅耗时耗力，还容易出错。尤其是当数据量达到 GB 级别时，手动处理几乎变得不可行。常见的痛点包括：

重复性操作多，如批量投影转换、裁剪、合并等
处理时间长，容易因人为失误中断
数据量大时，软件界面操作响应缓慢
难以保证处理过程的一致性

针对批量处理需求，主要有以下几种技术方案可选：

ArcPy：ArcGIS 自带的 Python 库，功能全面但仅限于桌面环境
ArcGIS API for Python：基于 Web GIS 场景，适合云端处理
GDAL/OGR：开源方案，跨平台但学习曲线较陡

对于初学者，建议从 ArcPy 开始，因为它：

直接集成在 ArcGIS Pro/Desktop 中
语法与 ArcGIS 操作高度对应
文档和社区资源丰富

使用 arcpy.da.Walk 可以高效遍历文件夹结构：

import arcpy
import os

workspace = r"C:\data"

# 遍历所有 shp 文件
for dirpath, dirnames, filenames in arcpy.da.Walk(workspace):
    for filename in filenames:
        if filename.endswith('.shp'):
            full_path = os.path.join(dirpath, filename)
            print(f"Processing: {full_path}")

批量处理前确保数据使用相同的空间参考：

# 定义目标坐标系
target_sr = arcpy.SpatialReference(4326)  # WGS84

# 批量投影转换
for shp in shapefiles:
    output = f"{shp}_projected"
    arcpy.Project_management(shp, output, target_sr)

利用 Python 的 multiprocessing 实现并行处理：

from multiprocessing import Pool
import functools

def process_file(input_file, output_folder):
    # 处理逻辑
    pass

if __name__ == '__main__':
    with Pool(processes=4) as pool:  # 使用 4 个进程
        pool.map(functools.partial(process_file, output_folder='output'), file_list)

完善的错误处理能让脚本更健壮：

import logging

logging.basicConfig(filename='batch_process.log', level=logging.ERROR)

try:
    # 处理代码
    arcpy.Buffer_analysis(input, output, distance)
except arcpy.ExecuteError as e:
    logging.error(f"处理 {input} 失败: {e}")
    continue

import arcpy
import os
import time
from tqdm import tqdm  # 进度条库

# 配置
input_folder = r"C:\data"
output_folder = r"C:\output"
sr = arcpy.SpatialReference(4326)  # WGS84

# 创建输出目录
if not os.path.exists(output_folder):
    os.makedirs(output_folder)

# 获取所有 shp 文件
shapefiles = []
for dirpath, _, filenames in arcpy.da.Walk(input_folder):
    for f in filenames:
        if f.endswith('.shp'):
            shapefiles.append(os.path.join(dirpath, f))

# 处理进度条
with tqdm(total=len(shapefiles)) as pbar:
    for shp in shapefiles:
        try:
            # 构造输出路径
            out_name = os.path.basename(shp).replace('.shp', '_processed.shp')
            out_path = os.path.join(output_folder, out_name)

            # 1. 投影转换
            projected = arcpy.Project_management(shp, "memory/temp", sr)

            # 2. 缓冲区分析
            arcpy.Buffer_analysis(projected, out_path, "100 Meters")

            # 更新进度
            pbar.update(1)
            pbar.set_description(f"处理: {out_name}")

        except Exception as e:
            arcpy.AddError(f"处理 {shp} 失败: {str(e)}")
            continue

print("批量处理完成！")