您没有发布
image1.tif
文件的链接,因此下面的示例代码使用
https://github.com/mdbartos/pysheds 中的
pysheds/data/dem.tif
。基本思路是将输入参数
xs
和
ys
在您的情况下分成子集,然后将每个 CPU 分配给不同的子集进行处理。
main()
会计算两次解决方案,一次按顺序执行,一次并行执行,然后比较每个解决方案。并行解决方案存在一些效率问题,因为每个 CPU 都将读取图像文件,因此有改进的空间(即在并行部分外部读取图像文件,然后将生成的
grid
对象提供给每个实例)。
import numpy as np
from pysheds.grid import Grid
from dask.distributed import Client
from dask import delayed, compute
xs = 10, 20, 30, 40, 50, 60, 70, 80, 90, 100
ys = 25, 35, 45, 55, 65, 75, 85, 95, 105, 115, 125
def var(image_file, x_in, y_in):
grid = Grid.from_raster(image_file, data_name='map')
variable_avg = []
for (x,y) in zip(x_in,y_in):
grid.catchment(data='map', x=x, y=y, out_name='catch')
variable = grid.view('catch', nodata=np.nan)
variable_avg.append( np.array(variable).mean() )
return(variable_avg)
def var_parallel(n_cpu, image_file, x_in, y_in):
tasks = []
for cpu in range(n_cpu):
x_in = xs[cpu::n_cpu]
y_in = ys[cpu::n_cpu]
tasks.append( delayed(var)(image_file, x_in, y_in) )
ans = compute(tasks)
par_avg = [None]*len(xs)
for cpu in range(n_cpu):
par_avg[cpu::n_cpu] = ans[0][cpu]
print('AVG (parallel) =',par_avg)
return par_avg
def main():
image_file = 'pysheds/data/dem.tif'
seq_avg = var(image_file, xs, ys)
print('AVG (sequential)=',seq_avg)
n_cpu = 3
dask_client = Client(n_workers=n_cpu)
par_avg = var_parallel(n_cpu, image_file, xs, ys)
dask_client.shutdown()
print('max error=',
max([ abs(seq_avg[i]-par_avg[i]) for i in range(len(seq_avg))]))
if __name__ == '__main__': main()
numba
。 - Mykola Zotko