使用python并行下载ERA5数据的方法

astiny

ERA5数据是我们常用的再分析资料之一。它的时空分辨率都比较高。


一般我们下载ERA5数据，可以直接在官网上，手动选择时间、变量等信息，然后点击链接下载；也可以使用ECMWF的Python cds API下载。（如http://bbs.06climate.com/forum.php?mod=viewthread&tid=91210）


multiprocessing包是Python中的多进程管理包。使用multiprocessing包进行并行下载，可以有效提高下载的速度。比如Gavin老师写的脚本（见http://climate2weather.cc/2019/05/10/era5-download/）


但该脚本也有一些小问题：cds的API仅支持最多12个任务（2020.2.17修改,已咨询ec并确认）同时提交。假如你同时提交多个下载进程，且其中后提交的进程下载速度比先提交的进程快，那么该进程结束，新的进程提交上去，会顶掉最早提交的那个进程，造成该进程无法顺利下载完毕。相信用过Gavin老师那个脚本的同学们，有些已经遇到这个问题了。那么，我们应该如何解决呢？


这个问题主要还是ERA5关于下载的介绍不完整所造成的。在官网的下载教程和实例代码中，仅介绍了retrieve这一种下载方法。这可能对大家造成了一种误解，以为只能使用这一种下载方法。其实不然。
熟悉python面向对象编程的同学们应该可以马上反应过来了：对于一个稍有编程经验的程序员来说，他开发时候，肯定是会把所有的方法（以ERA5的这个api为例，包括retrieve，download，delete等等）都开发在一个类里，不可能只提供一个下载的函数。EC的coding能力我还是比较信任的，他们应该干不出这种事情。


因此，我找到了EC在github上的项目https://github.com/ecmwf/cdsapi，然后在该repository中搜索"delete"。
果然，在https://github.com/ecmwf/cdsapi/blob/a656204578341cfc994a4772496b5aca1d823d49/cdsapi/api.py中，详细介绍了retrieve，download，delete等方法。在提交session然后download后，其实可以通过delete来删除你已经下载完的进程，这样就可以避免新的进程顶掉最早提交的进程了。

好了，说了那么多，一大半对于大多数同学都是没用的。还是代码见吧（基本上是从Gavin老师脚本上改的。运行到现在没出现过任何问题）。这里默认大家都懂怎么用python非并行下载了。不太懂的可以参考上面几个链接。

#############start################

1. from queue import Queue
   
2. from threading import Thread
   
3. import cdsapi
   
4. import time
   
5. from datetime import datetime, timedelta
   
6. import os, glob
   
7. from multiprocessing import Pool
   
8. 
   
9. download_dir = os.getcwd()
   
10. 
    
11. 
    
12. def download_ear5(allvar):
    
13.     print(allvar)
    
14.     variable = allvar[0]
    
15.     pressure_level = allvar[1]
    
16.     year = allvar[2]
    
17.     month = allvar[3]
    
18.     day = allvar[4]
    
19.     hour = allvar[5]
    
20.     file_path = os.path.join(
    
21.         download_dir,
    
22.         variable,
    
23.         pressure_level,
    
24.         year,
    
25.         month,
    
26.         day,
    
27.         variable
    
28.         + "_"
    
29.         + pressure_level
    
30.         + "hpa"
    
31.         + "_"
    
32.         + year
    
33.         + month
    
34.         + day
    
35.         + hour
    
36.         + ".nc",
    
37.     )
    
38.     if not os.path.exists(
    
39.         file_path
    
40.     ):  #  or (os.path.getsize(file_path) < 1024*1024*1.8) :
    
41.         c = cdsapi.Client()
    
42.         r = c.retrieve(
    
43.             "reanalysis-era5-pressure-levels",
    
44.             {
    
45.                 "product_type": "reanalysis",
    
46.                 "format": "netcdf",
    
47.                 "variable": [variable],
    
48.                 "pressure_level": [pressure_level],
    
49.                 "year": year,
    
50.                 "month": [month],
    
51.                 "day": [day],
    
52.                 "time": [hour + ":00"],
    
53.                 "area": [90, 0, 0, 180],
    
54.             },
    
55.         )
    
56.         r.download(
    
57.             os.path.join(
    
58.                 download_dir,
    
59.                 variable,
    
60.                 pressure_level,
    
61.                 year,
    
62.                 month,
    
63.                 day,
    
64.                 variable
    
65.                 + "_"
    
66.                 + pressure_level
    
67.                 + "hpa"
    
68.                 + "_"
    
69.                 + year
    
70.                 + month
    
71.                 + day
    
72.                 + hour
    
73.                 + ".nc",
    
74.             )
    
75.         )
    
76.         r.delete()
    
77.         time.sleep(3)
    
78.     return
    
79. 
    
80. 
    
81. # 下载脚本
    
82. class DownloadWorker(Thread):
    
83.     def __init__(self, queue):
    
84.         Thread.__init__(self)
    
85.         self.queue = queue
    
86. 
    
87.     def run(self):
    
88.         while True:
    
89.             # 从队列中获取任务并扩展tuple
    
90.             allvar = self.queue.get()
    
91.             download_ear5(allvar)
    
92.             self.queue.task_done()
    
93. 
    
94. 
    
95. # 主程序
    
96. def main():
    
97.     start_time = datetime(2017, 1, 1, 0)  # 开始时间
    
98.     end_time = datetime(2020, 1, 1, 0)  # 结束时间
    
99.     # 创建列表，存储起止时间之间的所有时次
    
100.     datetime_list = []
     
101.     while start_time <= end_time:
     
102.         datetime_list.append(start_time)
     
103.         start_time += timedelta(hours=1)
     
104.     # 变量列表
     
105.     variable_list = [
     
106.         "geopotential",
     
107.         "specific_humidity",
     
108.         "temperature",
     
109.         "u_component_of_wind",
     
110.         "v_component_of_wind",
     
111.     ]
     
112.     # 气压层列表
     
113.     pressure_level_list = ["100", "200", "500", "700", "850", "925", "1000"]
     
114.     # 创建allvar列表，内含variable,pressure_level,year,month,day,hour
     
115.     allvar = []
     
116.     for idatetime in datetime_list:  # 非并行创建文件夹，同事填充allvar列表
     
117.         year = idatetime.strftime("%Y")
     
118.         month = idatetime.strftime("%m")
     
119.         day = idatetime.strftime("%d")
     
120.         hour = idatetime.strftime("%H")
     
121.         for pressure_level in pressure_level_list:
     
122.             for variable in variable_list:
     
123.                 if not os.path.exists(
     
124.                     os.path.join(
     
125.                         download_dir, variable, pressure_level, year, month, day
     
126.                     )
     
127.                 ):
     
128.                     os.makedirs(
     
129.                         os.path.join(
     
130.                             download_dir, variable, pressure_level, year, month, day
     
131.                         )
     
132.                     )
     
133.                 allvar.append([variable, pressure_level, year, month, day, hour])
     
134. 
     
135.     # 创建一个主进程与工作进程通信
     
136.     queue = Queue()
     
137. 
     
138.     # 注意，每个用户同时最多接受4个request https://cds.climate.copernicus.eu/vision
     
139.     # 创建12个工作线程
     
140.     for x in range(12):
     
141.         worker = DownloadWorker(queue)
     
142.         # 将daemon设置为True将会使主线程退出，即使所有worker都阻塞了
     
143.         worker.daemon = False
     
144.         worker.start()
     
145. 
     
146.     # 将任务以tuple的形式放入队列中
     
147.     for link in allvar:
     
148.         queue.put((link))
     
149. 
     
150.     # 让主线程等待队列完成所有的任务
     
151.     queue.join()
     
152. 
     
153. 
     
154. if __name__ == "__main__":
     
155.     main()

复制代码

#############end###################

接下来是感想和杂谈时间。
1.不能只看教程。还是得养成多看源代码的习惯。不然，只能永远做一个调包侠，coding的技能没法成长。

2.我这里写的脚本，是将每一层、每个变量，单独存成一个文件。这样有一个好处：当你需要用部分数据的时候，不需要读取全部，节省了很多时间。有点和mongodb的感觉一样（笑）。但是，据我在许多气象相关微信群里的经验，有很多同学似乎不太适应用循环的方法读取多个文件。这也许是收了ncl的影响，因为ncl中的addfiles函数，提供了读取整个文件列表中类似格式的nc文件的方法。这固然提供了很多方便，但也许也使得大家下意识地回避了循环这个问题。实际上，我们不仅不应该逃避循环结构，而应该进一步地拥抱循环语句。我相信，对于熟悉数据结构和循环的同学来说，循环应该是一个十分得力的工具。如果不熟悉的话，希望大家也都熟悉起来，因为掌握循环和数据结构（不论对哪种语言来说），都是十分有必要的。

梁的丰

厉害，如何提高coding水平？

astiny

梁的丰 发表于 2020-2-16 17:44
厉害，如何提高coding水平？

个人经验是，多写多看，面向工程。没太多捷径。

平流层的萝卜

一直想研究一下并行下载！非常感谢楼主的分享，下了不少功夫！学习一下！

Alexander

强的一笔，学习学习{:eb302:}{:eb302:}{:eb302:}{:eb302:}{:eb302:}{:eb302:}

cxbhero

感谢分享！老师，请教个问题，ERA5数据也包含预报数据下载吧？如何下载呢，我想下载有效波高这类海洋数据！

Adiligirl

感谢LZ分享，学习到了

astiny

cxbhero 发表于 2020-2-19 22:33
感谢分享！老师，请教个问题，ERA5数据也包含预报数据下载吧？如何下载呢，我想下载有效波高这类海洋数据！

这是再分析呀。。。预报数据只有气象局有。要钱的。

chh

太强了！！！

chh

请问下载中间提示的下载不完全错误，后面会接着重新下载完成吗