Как оптимизировать файлы NetCDF и dask для обработки долгосрочных климатологических индексов с помощью xclim (например, ⇐ Python

[Anonymous]

Я пытаюсь проанализировать 30-дневный стандартизированный индекс осадков для диапазона нескольких штатов на юго-востоке США за 2016 год. Я использую xclim для обработки прямого получения данных о ежедневных осадках с координатной сеткой из nClimGrid-Daily, что включает в себя получение данных за около 40 лет и выполнение 30-дневного скользящего среднего значения суммы осадков за день. Разрешение суточное и 48,3 км. Я ожидал, что код предоставит мне 30-дневный SPI для всего диапазона, который я затем смогу подмножество на 2016 год. Вместо этого я запущу его на HPC моей школы через ssh, а затем получу половину вычислительного узла (вычислительный узел имеет около 70 ГБ ОЗУ). Я могу запустить его на своем локальном компьютере, но он слишком сильно нагревается.
Я загрузил необходимый мне диапазон (8,8 ГБ). После географических и временных подстановок это около 7 ГБ. Я могу лениво загружать все, но когда я сохраняю/вычисляю с помощью dask, он просто продолжает зависать после работы в течение определенного времени - большая часть вычислительной мощности уходит на перегруппировку с помощью xclim, хотя я пытался перегруппировать до этого, чтобы избежать этого. Обратите внимание: когда я использую более грубое пространственное разрешение (на уровне округа или в координатной сетке), я могу выполнить это менее чем за час. С решеткой он ломается примерно через 3 часа. Пожалуйста, дайте мне знать, если у вас есть какие-либо предложения!
import xarray as xr
from dask.distributed import Client, LocalCluster
import os
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

#### bulk download files
base_url = "https://www.ncei.noaa.gov/data/nclimgri ... cess/grids"
years = range(1976, 2017)
months = range(1, 13)

out_dir = Path("./nclimgrid_prcp") # adjust as needed
out_dir.mkdir(parents=True, exist_ok=True)

max_workers = 16
timeout = 60 # seconds

def dl(year, month):
fname = f"ncdd-{year}{month:02}-grd-scaled.nc"
url = f"{base_url}/{year}/{fname}"
path = out_dir / fname

if path.exists():
return path, "exists"

try:
r = requests.get(url, timeout=timeout, stream=True)
r.raise_for_status()
with open(path, "wb") as f:
for chunk in r.iter_content(chunk_size=2**20):
if chunk:
f.write(chunk)
return path, "downloaded"
except Exception as e:
return path, f"error: {e}"

futures = []
with ThreadPoolExecutor(max_workers=max_workers) as ex:
for y in years:
for m in months:
futures.append(ex.submit(dl, y, m))

files = []
for fut in as_completed(futures):
p, status = fut.result()
print(p.name, status)
if status in ("exists", "downloaded"):
files.append(str(p))

####### End download files

files = sorted(glob.glob('nclimgrid_prcp/*.nc')) #access local files

def subset(ds):
return ds.sel(
lat=slice(30, 38),
lon=slice(-92, -78),
)
ds = xr.open_mfdataset(files, combine="by_coords" preprocess=subset,engine='netcdf4')

pr = ds['prcp']
pr = pr.assign_attrs(units="mm/day")

pr = pr.chunk(
{
"time": -1,
"lat": 64,
"lon": 84,
}
)

cluster = LocalCluster()
client = Client(cluster)
client.cluster.scale(10)

spi_30 = xc.indices.standardized_precipitation_index(
pr,
freq="D",
window=30)


Подробнее здесь: https://stackoverflow.com/questions/798 ... ological-i