1. 从“看”到“拿”为什么我们需要批量下载与处理上一篇文章我们聊了怎么用Python和GEE搭上线还成功下载了一小块DEM数据。很多朋友试了之后跑来问我“老师这个例子我跑通了但我就拿到一个几十兆的zip包这离我实际项目需求差得有点远啊” 这话说到点子上了。第一次成功调用API看到数据下载到本地那种兴奋感我懂但这就像你刚学会开车只是在停车场转了两圈。真正的挑战是你需要开车去一个陌生的城市连续跑好几个地方还得把一路上拍的照片都整理好。科研和工程中的遥感数据需求往往就是这么个“长途多任务”场景。想象一下这些真实情况你的研究区是整个长江流域你需要过去10年每两个月一次的Landsat影像用来分析植被变化或者你要做一个全国性的城市扩张研究需要下载覆盖所有省会城市的夜间灯光数据又或者你接了一个环保项目要监测某个区域过去5年每个季度的水体面积。这些需求共同的特点是区域大、时间序列长、数据量巨大。如果你还按照我们上篇文章的方法手动改坐标、一遍遍运行脚本、下载完再手动解压拼接那估计项目还没做完人先崩溃了。这就是我们这次要解决的核心痛点如何把一次性的“手动拿”变成系统性的“自动搬”。GEE云端有海量数据计算能力也强大但它最终提供给你的往往是一个个数据“包裹”。我们的工作就是设计一套自动化流水线高效、准确、稳定地把这些“包裹”搬运到本地并初步整理成我们能用的格式。这个过程会涉及到几个关键环节如何自动遍历所有需要的时间和空间范围如何管理成百上千个下载任务不混乱下载下来的原始数据怎么快速进行质量检查和预处理这正是“批量下载与处理”要教给你的核心技能。掌握了它你才算是真正把GEE这个“数据宝库”的钥匙握在了自己手里可以随时按需取用而不再是面对宝库望洋兴叹。2. 搭建你的自动化流水线核心思路与工具准备搞批量处理最怕的就是思路不清代码写着写着就成了一团乱麻。在我多年的实战里总结了一个非常管用的核心心法“任务清单”驱动。别一上来就埋头写for循环我们先得把“要干什么”这件事梳理得明明白白。2.1 理解“任务”的构成时间、空间与数据产品一个最基本的遥感数据下载任务其实就由三个维度决定空间范围你要下载哪块地可能是一个省、一条河流或者几百个分散的采样点。时间范围你要哪个时间段的数据可能是连续的10年也可能是特定的几个月份。数据产品你要哪种卫星、哪种处理级别的影像比如Landsat 8的地表反射率数据LANDSAT/LC08/C02/T1_L2或者MODIS的植被指数数据MODIS/006/MOD13Q1。我们的自动化脚本本质上就是根据你定义好的空间列表、时间列表和数据产品列表自动生成一大堆具体的下载任务然后有条不紊地去执行。这里我强烈推荐你在写代码前先用纸笔或Excel表格把你的任务清单列出来。比如任务ID区域名称几何类型坐标/矢量文件开始日期结束日期数据产品目标分辨率1北京城区矩形[116.1, 39.8, 116.5, 40.1]2022-01-012022-12-31LANDSAT/LC08/C02/T1_L230米2太湖流域矢量文件taihu.shp2021-06-012021-09-01COPERNICUS/S2_SR10米有了这样一张表你的代码逻辑就会清晰无比读取表格的每一行解析参数构造GEE请求发起下载。2.2 本地环境与工具升级告别单打独斗上篇文章我们用了最基本的urllib来下载对于批量任务它有点力不从心。我们需要引入更强大的“帮手”。首先我强烈建议你安装geemap这个Python库。它不是GEE官方出的但可以说是Python版GEE的“瑞士军刀”由华人科学家吴秋生博士开发功能极其强大。pip install geemapgeemap不仅能更方便地可视化数据、交互式选择区域它内置的批量下载工具geemap.download_ee_image_collection能让我们省下大量造轮子的时间。其次对于下载环节我们用requests库替代urllib它的重试机制、会话管理和异常处理更完善。同时面对成百上千个文件我们需要zipfile库来解压os和pathlib库来管理本地复杂的文件夹结构。最后一个经常被忽略但至关重要的工具日志系统。批量任务一跑可能就是几个小时甚至几天没有日志你根本不知道它卡在哪了。Python自带的logging模块就很好用配置一下让程序把每个任务开始、成功、失败的信息都记录到文件里这样你随时可以查看进度排查问题。把这些工具准备好你的代码“工具箱”就升级完成了。接下来我们进入实战环节看看怎么用它们来构建流水线。3. 实战第一步批量构建与筛选你的数据列表现在我们开始动手写代码。假设我们的任务是下载中国东部五个主要城市群2023年全年云量低于10%的Landsat 8影像。3.1 定义多区域与时间序列我们首先把任务参数定义清楚。这里我用字典列表来模拟前面说的“任务清单”import ee import geemap import logging from datetime import datetime # 初始化GEE和日志 ee.Initialize() logging.basicConfig(filenamebatch_download.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) # 1. 定义我们的多个研究区域这里用矩形简单示例实际可用矢量文件导入 regions { Beijing: ee.Geometry.Rectangle([115.5, 39.5, 117.5, 40.5]), Shanghai: ee.Geometry.Rectangle([120.8, 30.7, 122.2, 31.8]), Guangzhou: ee.Geometry.Rectangle([112.9, 22.5, 114.1, 23.8]), Chengdu: ee.Geometry.Rectangle([103.5, 30.3, 104.5, 31.0]), Wuhan: ee.Geometry.Rectangle([113.8, 30.3, 114.8, 30.9]) } # 2. 定义时间范围 start_date 2023-01-01 end_date 2023-12-31 # 3. 定义数据产品 dataset_id LANDSAT/LC08/C02/T1_L23.2 利用GEE的筛选器高效获取影像集接下来我们不是一个个区域去循环而是先利用GEE强大的服务器端过滤能力一次性获取满足我们条件的所有影像集合。这是提升效率的关键因为过滤发生在谷歌服务器上比我们把数据下载到本地再筛选快无数倍。# 4. 加载影像集并应用时间和空间筛选这里以北京区域为例演示 collection ee.ImageCollection(dataset_id) \ .filterBounds(regions[Beijing]) \ .filterDate(start_date, end_date) \ .filter(ee.Filter.lt(CLOUD_COVER, 10)) # 筛选云量低于10%的影像 # 看看我们筛选出了多少景影像 print(f北京区域符合条件的影像数量{collection.size().getInfo()}) logging.info(f北京区域筛选完成共{collection.size().getInfo()}景影像。)这里有个非常重要的技巧collection.size().getInfo()这个操作会触发一次客户端与服务器的通信获取一个简单的数字。在批量任务前先用这个方式检查每个区域的数据量可以避免你兴冲冲地开始下载结果发现某个时间段根本没数据白白浪费感情。3.3 处理时间序列与生成下载列表如果数据量很大我们可能还需要按月份或者季度来分批下载。我们可以用时间循环来生成更精细的任务列表import calendar from dateutil.relativedelta import relativedelta def generate_monthly_intervals(start_str, end_str): 生成每月的起始和结束日期列表 start datetime.strptime(start_str, %Y-%m-%d) end datetime.strptime(end_str, %Y-%m-%d) intervals [] current start while current end: month_end current relativedelta(day31) # 获取当月最后一天 if month_end end: month_end end intervals.append(( current.strftime(%Y-%m-%d), month_end.strftime(%Y-%m-%d) )) current month_end relativedelta(days1) return intervals # 为北京区域生成月度下载任务 monthly_tasks [] for month_start, month_end in generate_monthly_intervals(start_date, end_date): monthly_collection collection.filterDate(month_start, month_end) count monthly_collection.size().getInfo() if count 0: # 只有当月有数据时才加入任务列表 monthly_tasks.append({ region: Beijing, geometry: regions[Beijing], start: month_start, end: month_end, collection: monthly_collection, expected_count: count }) logging.info(f生成任务北京{month_start} 至 {month_end}预计{count}景影像。)通过这样的步骤我们就把一个模糊的“下载全年数据”需求转化成了一个具体的、可执行的“任务对象”列表。每个任务对象包含了下载所需的所有信息。这才是自动化批量处理的正确起点。4. 核心引擎稳定可靠的批量下载与本地管理任务清单准备好了现在我们来打造下载引擎。这一步的核心目标是稳定、不断线、有条理。4.1 封装健壮的下载函数直接使用getDownloadUrl和简单下载在网络波动或数据过大时很容易失败。我们必须给它加上“盔甲”。import requests import time from pathlib import Path def download_gee_image(image, geometry, scale, crs, region_name, date_str, base_dir./downloads): 下载单景GEE影像包含重试机制和详细日志。 参数: image: ee.Image对象 geometry: 下载区域几何体 scale: 分辨率 crs: 坐标系 region_name: 区域名称 date_str: 日期标识字符串 base_dir: 本地存储根目录 # 1. 创建有结构的本地文件夹 save_dir Path(base_dir) / region_name / date_str[:7] # 按区域/年月组织 save_dir.mkdir(parentsTrue, exist_okTrue) filename f{region_name}_{date_str}_{scale}m.tif filepath save_dir / filename # 如果文件已存在跳过下载 if filepath.exists(): logging.warning(f文件已存在跳过{filepath}) return str(filepath) # 2. 获取下载链接 try: download_url image.getDownloadURL({ scale: scale, crs: crs, region: geometry, format: GEO_TIFF # 直接下载为GeoTIFF省去解压zip步骤 }) except Exception as e: logging.error(f获取下载链接失败{region_name}, {date_str}。错误{e}) return None # 3. 带重试机制的下载 max_retries 3 for attempt in range(max_retries): try: logging.info(f开始下载 [{attempt1}/{max_retries}]{filename}) # 使用requests库设置流式下载和超时 response requests.get(download_url, streamTrue, timeout60) response.raise_for_status() # 检查HTTP错误 # 分块写入文件避免大文件内存溢出 with open(filepath, wb) as f: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) logging.info(f下载成功{filepath}) return str(filepath) except requests.exceptions.RequestException as e: logging.warning(f下载尝试 {attempt1} 失败{filename}。错误{e}) if attempt max_retries - 1: wait_time (attempt 1) * 10 # 重试等待时间递增 logging.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: logging.error(f下载最终失败已重试{max_retries}次{filename}) return None这个函数做了几件关键事结构化存储文件不乱、断点续传通过检查文件是否存在、异常捕获与重试网络不稳定也不怕、流式下载大文件无忧。这些都是批量下载中血泪教训换来的经验。4.2 调度与遍历整个任务列表有了健壮的下载单元我们就可以用循环来调度整个任务列表了。这里需要注意加入任务间的延迟避免对GEE服务器请求过于频繁导致被临时限制。def process_task_batch(task_list, scale30, crsEPSG:4326): 处理一个任务批次 successful_downloads [] failed_downloads [] for i, task in enumerate(task_list): region_name task[region] geometry task[geometry] start_date task[start] logging.info(f开始处理任务 {i1}/{len(task_list)}: {region_name} - {start_date}) # 这里需要根据实际情况从collection中获取单景影像。 # 例如我们可以选择云量最低的一景或者进行中值合成。 # 以下以选择collection中第一景为例实际应用需更精细的选择逻辑 image_collection task[collection] try: # 获取影像集的第一景影像 image ee.Image(image_collection.first()) # 获取该影像的日期属性作为标识 date_str ee.Date(image.get(system:time_start)).format(YYYY-MM-dd).getInfo() except Exception as e: logging.error(f从集合中获取影像失败{region_name}, {task[start]}。错误{e}) failed_downloads.append(task) continue # 调用下载函数 result_path download_gee_image( imageimage, geometrygeometry, scalescale, crscrs, region_nameregion_name, date_strdate_str ) if result_path: successful_downloads.append((region_name, date_str, result_path)) else: failed_downloads.append(task) # 任务间短暂暂停友好访问 time.sleep(2) # 输出批次报告 logging.info(f批次处理完成。成功{len(successful_downloads)}失败{len(failed_downloads)}) return successful_downloads, failed_downloads # 执行我们之前生成的月度任务 success, fail process_task_batch(monthly_tasks)通过这样的架构你的批量下载任务就从一个脆弱的脚本变成了一个具有生产级鲁棒性的小型系统。它可以处理中断可以记录详尽的日志可以优雅地重试并且把所有数据分门别类地保存好。5. 数据到手不是终点自动化预处理与质量检查数据下载到本地硬盘只是完成了第一步。一堆原始的GeoTIFF文件可能还存在一些问题比如不同景影像之间因为采集时间不同存在色差或者某些区域被云层遮盖又或者你需要把它们全部裁剪到完全相同的范围并镶嵌起来。这些工作如果手动在QGIS或ArcGIS里操作工作量不可想象。我们需要让预处理也自动化起来。5.1 使用Rasterio进行批量基础处理rasterio是一个处理栅格数据非常强大的Python库。我们可以用它来写一些脚本对下载好的整个文件夹进行批量操作。示例1批量重投影和裁剪。假设我们所有影像需要统一到UTM投影并且裁剪到我们研究区域的精确矢量边界。import rasterio from rasterio.mask import mask import geopandas as gpd from pathlib import Path def batch_reproject_and_clip(input_dir, output_dir, shapefile_path, target_crsEPSG:32650): 批量将某个文件夹下的TIFF文件重投影并裁剪到矢量边界。 input_dir Path(input_dir) output_dir Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) # 读取用于裁剪的矢量边界 gdf gpd.read_file(shapefile_path) # 确保矢量边界与目标CRS一致如果不同则转换 if gdf.crs ! target_crs: gdf gdf.to_crs(target_crs) # 获取几何形状取第一个要素 geom gdf.geometry.iloc[0] for tif_file in input_dir.glob(*.tif): output_file output_dir / fprocessed_{tif_file.name} with rasterio.open(tif_file) as src: # 1. 计算重投影后的变换参数和图像尺寸 transform, width, height rasterio.warp.calculate_default_transform( src.crs, target_crs, src.width, src.height, *src.bounds) kwargs src.meta.copy() kwargs.update({ crs: target_crs, transform: transform, width: width, height: height }) # 2. 执行重投影 reprojected_data rasterio.warp.reproject( sourcerasterio.band(src, 1), destinationrasterio.band(dst, 1), src_transformsrc.transform, src_crssrc.crs, dst_transformtransform, dst_crstarget_crs, resamplingrasterio.warp.Resampling.nearest) # 3. 用重投影后的数据创建临时文件然后进行裁剪这里简化流程实际需分步或使用内存文件 # 为简化示例此处跳过中间步骤。实际应用中你可能需要先保存重投影结果再加载进行裁剪。 # 更高效的做法是使用rasterio的虚拟文件或内存文件处理。 logging.info(f已处理{tif_file.name})示例2批量计算NDVI并输出。如果你下载的是多波段影像如Landsat预处理可能还包括指数计算。def batch_calculate_ndvi(image_dir, red_band_idx4, nir_band_idx5): 批量计算文件夹中影像的NDVI for tif_file in Path(image_dir).glob(*.tif): with rasterio.open(tif_file) as src: red src.read(red_band_idx).astype(float) nir src.read(nir_band_idx).astype(float) # 避免除零错误 ndvi (nir - red) / (nir red 1e-10) # 保存NDVI结果 profile src.profile profile.update(dtyperasterio.float32, count1) output_path tif_file.parent / fndvi_{tif_file.stem}.tif with rasterio.open(output_path, w, **profile) as dst: dst.write(ndvi.astype(rasterio.float32), 1) print(f已生成NDVI: {output_path})5.2 集成到流水线中最理想的状态是下载任务完成后自动触发预处理脚本。我们可以在主下载函数成功后调用这些处理函数或者设置一个独立的“后处理”调度程序定期扫描下载目录处理新文件。# 在主下载循环结束后可以这样调用 successful_files [item[2] for item in success] # 获取成功下载的文件路径列表 if successful_files: # 假设我们有一个统一的矢量边界文件 shapefile ./study_area/boundary.shp # 为每个文件所在的目录进行批量处理实际需根据组织方式调整 processed_dir ./processed_data for file_path in successful_files: input_dir Path(file_path).parent batch_reproject_and_clip(input_dir, processed_dir, shapefile) logging.info(所有下载数据预处理完成。)5.3 质量检查自动化批量处理容易“垃圾进垃圾出”。我们需要一个快速检查机制确保下载和处理的文件是有效的。一个简单的检查脚本可以包括文件是否能正常用rasterio.open打开。影像的CRS、尺寸是否符合预期。数据值是否在合理范围内比如NDVI应在-1到1之间。是否存在全是空值Nodata的无效文件。把这些检查点写成函数在关键步骤后运行就能及时发现问题避免错误累积到后期无法收拾。6. 避坑指南与高级技巧让流水线坚如磐石搞了这么多年自动化我踩过的坑比下载的数据都多。下面这些经验希望能帮你少走弯路。坑1配额超限与请求限制。GEE对每个用户的导出请求有配额限制。如果你一次性提交几百个导出任务很可能会被拒绝。解决方案一是错峰在脚本里加入随机延时把任务分散到几个小时甚至几天内完成二是优先使用getDownloadURL进行小范围、高分辨率数据的直接下载它消耗的配额比创建导出任务 (Export.image.toDrive) 要少。对于超大范围、全分辨率的数据导出到Google Drive是唯一选择那就需要精心规划分批分天进行。坑2几何区域过大或过于复杂。当你定义的下载区域尤其是复杂多边形顶点太多时GEE的getDownloadURL可能会报错或返回空数据。解决方案对于大区域先用geometry.bounds()获取其外接矩形进行下载下载到本地后再用GIS软件或rasterio进行精确裁剪。对于复杂多边形可以尝试用geometry.simplify()适当简化几何形状在不影响精度的前提下减少顶点数。坑3网络不稳定与断点续传。这是最常遇到的问题。我们前面已经在下载函数里加入了重试机制。更进一步你可以考虑使用更专业的下载工具如wget通过subprocess调用或aria2它们对断点续传的支持更原生。另外务必用好日志记录每个文件的下载状态未开始、下载中、成功、失败这样即使程序中断重启后也能从上次失败的地方继续。坑4数据产品选择与参数理解。GEE上同一个卫星可能有多个数据产品如地表反射率、大气顶部反射率、各种指数产品。选错了产品后续分析全白做。高级技巧在写死数据ID之前一定要去GEE的官方数据目录在Code Editor的Assets标签页里仔细阅读该数据集的文档搞清楚每个波段的含义、单位、缩放因子、以及云掩膜的方法。比如Landsat的SR地表反射率产品像素值通常乘以了0.0000275并加了-0.2的偏移使用前需要还原。坑5本地文件管理与元数据丢失。批量下载会产生大量文件如果没有良好的命名和组织规范很快就会变成一团乱麻。我的习惯是采用{区域代号}_{数据源}_{日期}_{分辨率}.tif这样的命名规则。同时强烈建议用一个CSV文件或SQLite数据库来管理下载任务的元数据任务ID、区域、时间、GEE使用的参数、本地存储路径、下载状态、文件哈希值用于校验完整性。这在你需要回溯或重新下载部分数据时能救命。最后分享一个我常用的高级模式“侦察兵”脚本。在启动庞大的正式下载任务前先运行一个轻量级的侦察脚本。这个脚本会遍历所有计划的任务只调用collection.size()和collection.first().getInfo()这类轻量操作快速检查每个时空单元是否有数据、数据量多大、大概的云覆盖情况。生成一份详细的侦察报告后你再根据报告调整你的任务清单比如跳过完全没有数据的月份这样可以极大提高正式运行的效率和成功率。磨刀不误砍柴工在批量处理的世界里前期规划的时间最终都会在后期调试和补救中加倍省回来。
Python+GEE实战：从零开始实现遥感影像批量下载与处理（GEE系列2）
1. 从“看”到“拿”为什么我们需要批量下载与处理上一篇文章我们聊了怎么用Python和GEE搭上线还成功下载了一小块DEM数据。很多朋友试了之后跑来问我“老师这个例子我跑通了但我就拿到一个几十兆的zip包这离我实际项目需求差得有点远啊” 这话说到点子上了。第一次成功调用API看到数据下载到本地那种兴奋感我懂但这就像你刚学会开车只是在停车场转了两圈。真正的挑战是你需要开车去一个陌生的城市连续跑好几个地方还得把一路上拍的照片都整理好。科研和工程中的遥感数据需求往往就是这么个“长途多任务”场景。想象一下这些真实情况你的研究区是整个长江流域你需要过去10年每两个月一次的Landsat影像用来分析植被变化或者你要做一个全国性的城市扩张研究需要下载覆盖所有省会城市的夜间灯光数据又或者你接了一个环保项目要监测某个区域过去5年每个季度的水体面积。这些需求共同的特点是区域大、时间序列长、数据量巨大。如果你还按照我们上篇文章的方法手动改坐标、一遍遍运行脚本、下载完再手动解压拼接那估计项目还没做完人先崩溃了。这就是我们这次要解决的核心痛点如何把一次性的“手动拿”变成系统性的“自动搬”。GEE云端有海量数据计算能力也强大但它最终提供给你的往往是一个个数据“包裹”。我们的工作就是设计一套自动化流水线高效、准确、稳定地把这些“包裹”搬运到本地并初步整理成我们能用的格式。这个过程会涉及到几个关键环节如何自动遍历所有需要的时间和空间范围如何管理成百上千个下载任务不混乱下载下来的原始数据怎么快速进行质量检查和预处理这正是“批量下载与处理”要教给你的核心技能。掌握了它你才算是真正把GEE这个“数据宝库”的钥匙握在了自己手里可以随时按需取用而不再是面对宝库望洋兴叹。2. 搭建你的自动化流水线核心思路与工具准备搞批量处理最怕的就是思路不清代码写着写着就成了一团乱麻。在我多年的实战里总结了一个非常管用的核心心法“任务清单”驱动。别一上来就埋头写for循环我们先得把“要干什么”这件事梳理得明明白白。2.1 理解“任务”的构成时间、空间与数据产品一个最基本的遥感数据下载任务其实就由三个维度决定空间范围你要下载哪块地可能是一个省、一条河流或者几百个分散的采样点。时间范围你要哪个时间段的数据可能是连续的10年也可能是特定的几个月份。数据产品你要哪种卫星、哪种处理级别的影像比如Landsat 8的地表反射率数据LANDSAT/LC08/C02/T1_L2或者MODIS的植被指数数据MODIS/006/MOD13Q1。我们的自动化脚本本质上就是根据你定义好的空间列表、时间列表和数据产品列表自动生成一大堆具体的下载任务然后有条不紊地去执行。这里我强烈推荐你在写代码前先用纸笔或Excel表格把你的任务清单列出来。比如任务ID区域名称几何类型坐标/矢量文件开始日期结束日期数据产品目标分辨率1北京城区矩形[116.1, 39.8, 116.5, 40.1]2022-01-012022-12-31LANDSAT/LC08/C02/T1_L230米2太湖流域矢量文件taihu.shp2021-06-012021-09-01COPERNICUS/S2_SR10米有了这样一张表你的代码逻辑就会清晰无比读取表格的每一行解析参数构造GEE请求发起下载。2.2 本地环境与工具升级告别单打独斗上篇文章我们用了最基本的urllib来下载对于批量任务它有点力不从心。我们需要引入更强大的“帮手”。首先我强烈建议你安装geemap这个Python库。它不是GEE官方出的但可以说是Python版GEE的“瑞士军刀”由华人科学家吴秋生博士开发功能极其强大。pip install geemapgeemap不仅能更方便地可视化数据、交互式选择区域它内置的批量下载工具geemap.download_ee_image_collection能让我们省下大量造轮子的时间。其次对于下载环节我们用requests库替代urllib它的重试机制、会话管理和异常处理更完善。同时面对成百上千个文件我们需要zipfile库来解压os和pathlib库来管理本地复杂的文件夹结构。最后一个经常被忽略但至关重要的工具日志系统。批量任务一跑可能就是几个小时甚至几天没有日志你根本不知道它卡在哪了。Python自带的logging模块就很好用配置一下让程序把每个任务开始、成功、失败的信息都记录到文件里这样你随时可以查看进度排查问题。把这些工具准备好你的代码“工具箱”就升级完成了。接下来我们进入实战环节看看怎么用它们来构建流水线。3. 实战第一步批量构建与筛选你的数据列表现在我们开始动手写代码。假设我们的任务是下载中国东部五个主要城市群2023年全年云量低于10%的Landsat 8影像。3.1 定义多区域与时间序列我们首先把任务参数定义清楚。这里我用字典列表来模拟前面说的“任务清单”import ee import geemap import logging from datetime import datetime # 初始化GEE和日志 ee.Initialize() logging.basicConfig(filenamebatch_download.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) # 1. 定义我们的多个研究区域这里用矩形简单示例实际可用矢量文件导入 regions { Beijing: ee.Geometry.Rectangle([115.5, 39.5, 117.5, 40.5]), Shanghai: ee.Geometry.Rectangle([120.8, 30.7, 122.2, 31.8]), Guangzhou: ee.Geometry.Rectangle([112.9, 22.5, 114.1, 23.8]), Chengdu: ee.Geometry.Rectangle([103.5, 30.3, 104.5, 31.0]), Wuhan: ee.Geometry.Rectangle([113.8, 30.3, 114.8, 30.9]) } # 2. 定义时间范围 start_date 2023-01-01 end_date 2023-12-31 # 3. 定义数据产品 dataset_id LANDSAT/LC08/C02/T1_L23.2 利用GEE的筛选器高效获取影像集接下来我们不是一个个区域去循环而是先利用GEE强大的服务器端过滤能力一次性获取满足我们条件的所有影像集合。这是提升效率的关键因为过滤发生在谷歌服务器上比我们把数据下载到本地再筛选快无数倍。# 4. 加载影像集并应用时间和空间筛选这里以北京区域为例演示 collection ee.ImageCollection(dataset_id) \ .filterBounds(regions[Beijing]) \ .filterDate(start_date, end_date) \ .filter(ee.Filter.lt(CLOUD_COVER, 10)) # 筛选云量低于10%的影像 # 看看我们筛选出了多少景影像 print(f北京区域符合条件的影像数量{collection.size().getInfo()}) logging.info(f北京区域筛选完成共{collection.size().getInfo()}景影像。)这里有个非常重要的技巧collection.size().getInfo()这个操作会触发一次客户端与服务器的通信获取一个简单的数字。在批量任务前先用这个方式检查每个区域的数据量可以避免你兴冲冲地开始下载结果发现某个时间段根本没数据白白浪费感情。3.3 处理时间序列与生成下载列表如果数据量很大我们可能还需要按月份或者季度来分批下载。我们可以用时间循环来生成更精细的任务列表import calendar from dateutil.relativedelta import relativedelta def generate_monthly_intervals(start_str, end_str): 生成每月的起始和结束日期列表 start datetime.strptime(start_str, %Y-%m-%d) end datetime.strptime(end_str, %Y-%m-%d) intervals [] current start while current end: month_end current relativedelta(day31) # 获取当月最后一天 if month_end end: month_end end intervals.append(( current.strftime(%Y-%m-%d), month_end.strftime(%Y-%m-%d) )) current month_end relativedelta(days1) return intervals # 为北京区域生成月度下载任务 monthly_tasks [] for month_start, month_end in generate_monthly_intervals(start_date, end_date): monthly_collection collection.filterDate(month_start, month_end) count monthly_collection.size().getInfo() if count 0: # 只有当月有数据时才加入任务列表 monthly_tasks.append({ region: Beijing, geometry: regions[Beijing], start: month_start, end: month_end, collection: monthly_collection, expected_count: count }) logging.info(f生成任务北京{month_start} 至 {month_end}预计{count}景影像。)通过这样的步骤我们就把一个模糊的“下载全年数据”需求转化成了一个具体的、可执行的“任务对象”列表。每个任务对象包含了下载所需的所有信息。这才是自动化批量处理的正确起点。4. 核心引擎稳定可靠的批量下载与本地管理任务清单准备好了现在我们来打造下载引擎。这一步的核心目标是稳定、不断线、有条理。4.1 封装健壮的下载函数直接使用getDownloadUrl和简单下载在网络波动或数据过大时很容易失败。我们必须给它加上“盔甲”。import requests import time from pathlib import Path def download_gee_image(image, geometry, scale, crs, region_name, date_str, base_dir./downloads): 下载单景GEE影像包含重试机制和详细日志。 参数: image: ee.Image对象 geometry: 下载区域几何体 scale: 分辨率 crs: 坐标系 region_name: 区域名称 date_str: 日期标识字符串 base_dir: 本地存储根目录 # 1. 创建有结构的本地文件夹 save_dir Path(base_dir) / region_name / date_str[:7] # 按区域/年月组织 save_dir.mkdir(parentsTrue, exist_okTrue) filename f{region_name}_{date_str}_{scale}m.tif filepath save_dir / filename # 如果文件已存在跳过下载 if filepath.exists(): logging.warning(f文件已存在跳过{filepath}) return str(filepath) # 2. 获取下载链接 try: download_url image.getDownloadURL({ scale: scale, crs: crs, region: geometry, format: GEO_TIFF # 直接下载为GeoTIFF省去解压zip步骤 }) except Exception as e: logging.error(f获取下载链接失败{region_name}, {date_str}。错误{e}) return None # 3. 带重试机制的下载 max_retries 3 for attempt in range(max_retries): try: logging.info(f开始下载 [{attempt1}/{max_retries}]{filename}) # 使用requests库设置流式下载和超时 response requests.get(download_url, streamTrue, timeout60) response.raise_for_status() # 检查HTTP错误 # 分块写入文件避免大文件内存溢出 with open(filepath, wb) as f: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) logging.info(f下载成功{filepath}) return str(filepath) except requests.exceptions.RequestException as e: logging.warning(f下载尝试 {attempt1} 失败{filename}。错误{e}) if attempt max_retries - 1: wait_time (attempt 1) * 10 # 重试等待时间递增 logging.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: logging.error(f下载最终失败已重试{max_retries}次{filename}) return None这个函数做了几件关键事结构化存储文件不乱、断点续传通过检查文件是否存在、异常捕获与重试网络不稳定也不怕、流式下载大文件无忧。这些都是批量下载中血泪教训换来的经验。4.2 调度与遍历整个任务列表有了健壮的下载单元我们就可以用循环来调度整个任务列表了。这里需要注意加入任务间的延迟避免对GEE服务器请求过于频繁导致被临时限制。def process_task_batch(task_list, scale30, crsEPSG:4326): 处理一个任务批次 successful_downloads [] failed_downloads [] for i, task in enumerate(task_list): region_name task[region] geometry task[geometry] start_date task[start] logging.info(f开始处理任务 {i1}/{len(task_list)}: {region_name} - {start_date}) # 这里需要根据实际情况从collection中获取单景影像。 # 例如我们可以选择云量最低的一景或者进行中值合成。 # 以下以选择collection中第一景为例实际应用需更精细的选择逻辑 image_collection task[collection] try: # 获取影像集的第一景影像 image ee.Image(image_collection.first()) # 获取该影像的日期属性作为标识 date_str ee.Date(image.get(system:time_start)).format(YYYY-MM-dd).getInfo() except Exception as e: logging.error(f从集合中获取影像失败{region_name}, {task[start]}。错误{e}) failed_downloads.append(task) continue # 调用下载函数 result_path download_gee_image( imageimage, geometrygeometry, scalescale, crscrs, region_nameregion_name, date_strdate_str ) if result_path: successful_downloads.append((region_name, date_str, result_path)) else: failed_downloads.append(task) # 任务间短暂暂停友好访问 time.sleep(2) # 输出批次报告 logging.info(f批次处理完成。成功{len(successful_downloads)}失败{len(failed_downloads)}) return successful_downloads, failed_downloads # 执行我们之前生成的月度任务 success, fail process_task_batch(monthly_tasks)通过这样的架构你的批量下载任务就从一个脆弱的脚本变成了一个具有生产级鲁棒性的小型系统。它可以处理中断可以记录详尽的日志可以优雅地重试并且把所有数据分门别类地保存好。5. 数据到手不是终点自动化预处理与质量检查数据下载到本地硬盘只是完成了第一步。一堆原始的GeoTIFF文件可能还存在一些问题比如不同景影像之间因为采集时间不同存在色差或者某些区域被云层遮盖又或者你需要把它们全部裁剪到完全相同的范围并镶嵌起来。这些工作如果手动在QGIS或ArcGIS里操作工作量不可想象。我们需要让预处理也自动化起来。5.1 使用Rasterio进行批量基础处理rasterio是一个处理栅格数据非常强大的Python库。我们可以用它来写一些脚本对下载好的整个文件夹进行批量操作。示例1批量重投影和裁剪。假设我们所有影像需要统一到UTM投影并且裁剪到我们研究区域的精确矢量边界。import rasterio from rasterio.mask import mask import geopandas as gpd from pathlib import Path def batch_reproject_and_clip(input_dir, output_dir, shapefile_path, target_crsEPSG:32650): 批量将某个文件夹下的TIFF文件重投影并裁剪到矢量边界。 input_dir Path(input_dir) output_dir Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) # 读取用于裁剪的矢量边界 gdf gpd.read_file(shapefile_path) # 确保矢量边界与目标CRS一致如果不同则转换 if gdf.crs ! target_crs: gdf gdf.to_crs(target_crs) # 获取几何形状取第一个要素 geom gdf.geometry.iloc[0] for tif_file in input_dir.glob(*.tif): output_file output_dir / fprocessed_{tif_file.name} with rasterio.open(tif_file) as src: # 1. 计算重投影后的变换参数和图像尺寸 transform, width, height rasterio.warp.calculate_default_transform( src.crs, target_crs, src.width, src.height, *src.bounds) kwargs src.meta.copy() kwargs.update({ crs: target_crs, transform: transform, width: width, height: height }) # 2. 执行重投影 reprojected_data rasterio.warp.reproject( sourcerasterio.band(src, 1), destinationrasterio.band(dst, 1), src_transformsrc.transform, src_crssrc.crs, dst_transformtransform, dst_crstarget_crs, resamplingrasterio.warp.Resampling.nearest) # 3. 用重投影后的数据创建临时文件然后进行裁剪这里简化流程实际需分步或使用内存文件 # 为简化示例此处跳过中间步骤。实际应用中你可能需要先保存重投影结果再加载进行裁剪。 # 更高效的做法是使用rasterio的虚拟文件或内存文件处理。 logging.info(f已处理{tif_file.name})示例2批量计算NDVI并输出。如果你下载的是多波段影像如Landsat预处理可能还包括指数计算。def batch_calculate_ndvi(image_dir, red_band_idx4, nir_band_idx5): 批量计算文件夹中影像的NDVI for tif_file in Path(image_dir).glob(*.tif): with rasterio.open(tif_file) as src: red src.read(red_band_idx).astype(float) nir src.read(nir_band_idx).astype(float) # 避免除零错误 ndvi (nir - red) / (nir red 1e-10) # 保存NDVI结果 profile src.profile profile.update(dtyperasterio.float32, count1) output_path tif_file.parent / fndvi_{tif_file.stem}.tif with rasterio.open(output_path, w, **profile) as dst: dst.write(ndvi.astype(rasterio.float32), 1) print(f已生成NDVI: {output_path})5.2 集成到流水线中最理想的状态是下载任务完成后自动触发预处理脚本。我们可以在主下载函数成功后调用这些处理函数或者设置一个独立的“后处理”调度程序定期扫描下载目录处理新文件。# 在主下载循环结束后可以这样调用 successful_files [item[2] for item in success] # 获取成功下载的文件路径列表 if successful_files: # 假设我们有一个统一的矢量边界文件 shapefile ./study_area/boundary.shp # 为每个文件所在的目录进行批量处理实际需根据组织方式调整 processed_dir ./processed_data for file_path in successful_files: input_dir Path(file_path).parent batch_reproject_and_clip(input_dir, processed_dir, shapefile) logging.info(所有下载数据预处理完成。)5.3 质量检查自动化批量处理容易“垃圾进垃圾出”。我们需要一个快速检查机制确保下载和处理的文件是有效的。一个简单的检查脚本可以包括文件是否能正常用rasterio.open打开。影像的CRS、尺寸是否符合预期。数据值是否在合理范围内比如NDVI应在-1到1之间。是否存在全是空值Nodata的无效文件。把这些检查点写成函数在关键步骤后运行就能及时发现问题避免错误累积到后期无法收拾。6. 避坑指南与高级技巧让流水线坚如磐石搞了这么多年自动化我踩过的坑比下载的数据都多。下面这些经验希望能帮你少走弯路。坑1配额超限与请求限制。GEE对每个用户的导出请求有配额限制。如果你一次性提交几百个导出任务很可能会被拒绝。解决方案一是错峰在脚本里加入随机延时把任务分散到几个小时甚至几天内完成二是优先使用getDownloadURL进行小范围、高分辨率数据的直接下载它消耗的配额比创建导出任务 (Export.image.toDrive) 要少。对于超大范围、全分辨率的数据导出到Google Drive是唯一选择那就需要精心规划分批分天进行。坑2几何区域过大或过于复杂。当你定义的下载区域尤其是复杂多边形顶点太多时GEE的getDownloadURL可能会报错或返回空数据。解决方案对于大区域先用geometry.bounds()获取其外接矩形进行下载下载到本地后再用GIS软件或rasterio进行精确裁剪。对于复杂多边形可以尝试用geometry.simplify()适当简化几何形状在不影响精度的前提下减少顶点数。坑3网络不稳定与断点续传。这是最常遇到的问题。我们前面已经在下载函数里加入了重试机制。更进一步你可以考虑使用更专业的下载工具如wget通过subprocess调用或aria2它们对断点续传的支持更原生。另外务必用好日志记录每个文件的下载状态未开始、下载中、成功、失败这样即使程序中断重启后也能从上次失败的地方继续。坑4数据产品选择与参数理解。GEE上同一个卫星可能有多个数据产品如地表反射率、大气顶部反射率、各种指数产品。选错了产品后续分析全白做。高级技巧在写死数据ID之前一定要去GEE的官方数据目录在Code Editor的Assets标签页里仔细阅读该数据集的文档搞清楚每个波段的含义、单位、缩放因子、以及云掩膜的方法。比如Landsat的SR地表反射率产品像素值通常乘以了0.0000275并加了-0.2的偏移使用前需要还原。坑5本地文件管理与元数据丢失。批量下载会产生大量文件如果没有良好的命名和组织规范很快就会变成一团乱麻。我的习惯是采用{区域代号}_{数据源}_{日期}_{分辨率}.tif这样的命名规则。同时强烈建议用一个CSV文件或SQLite数据库来管理下载任务的元数据任务ID、区域、时间、GEE使用的参数、本地存储路径、下载状态、文件哈希值用于校验完整性。这在你需要回溯或重新下载部分数据时能救命。最后分享一个我常用的高级模式“侦察兵”脚本。在启动庞大的正式下载任务前先运行一个轻量级的侦察脚本。这个脚本会遍历所有计划的任务只调用collection.size()和collection.first().getInfo()这类轻量操作快速检查每个时空单元是否有数据、数据量多大、大概的云覆盖情况。生成一份详细的侦察报告后你再根据报告调整你的任务清单比如跳过完全没有数据的月份这样可以极大提高正式运行的效率和成功率。磨刀不误砍柴工在批量处理的世界里前期规划的时间最终都会在后期调试和补救中加倍省回来。
