Я работаю над системой HPC ( Cheyenne * от NCAR ) и хочу сделать несколько вещей с большим набором данных NetCDF (Большой ансамбль NCAR).У меня проблемы с памятью, несмотря на чтение документации Pangeo и dask / dask.
Я хочу предпринять следующие шаги:
- Большая загрузка (глобальная,~ 1800 лет суточных данных), набор данных из нескольких файлов, используя
xarray.open_mfdataset() в сумме - Возьмите 30-дневную скользящую сумму по измерению полного времени во всех точках сетки, в идеале, используя распределенную dask, посколькуПорядок данных в ТБ
- Рассчитать интервалы возврата по этим 30-дневным суммам (т. е. запустить
numpy.nanpercentile() или функцию dask DataFrame.quantile() с указанным значением процентиля) - Сохранить результатв виде 2D (шир. и долг.) файла NetCDF
Для меня важно иметь возможность загружать полный временной ряд для любой точки сетки одновременно, потому что вычисления скользящей суммы и интервала возвратазависит от полного периода времени.
Вот фрагмент кода, с которым я пытался это сделать (до шага 2 выше).Я запускаю его в блокноте Jupyter, и я указал $TMPDIR в качестве своего личного каталога для заметок (в этой системе /glade/scratch/$USER/temp/):
import xarray
import dask_jobqueue
import dask.distributed
import glob
## specify location of precipitation files, use glob to get list of them
PRECT_root = '/glade/collections/cdg/data/cesmLE/CESM-CAM5-BGC-LE/atm/proc/tseries/daily/PRECT/'
filename_list = sorted(glob.glob(PRECT_root + 'b.e11.B1850C5CN.f09_g16.005.cam.h1.PRECT.*.nc'))
## import multi-file data set
mfds = xarray.open_mfdataset(filename_list, chunks={'lat':12, 'lon':12})
mfds_data = mfds['PRECT']
## start cluster object
cluster = dask_jobqueue.PBSCluster(cores=36, memory='100 GB', project='UCLA0022', queue='regular', walltime='06:00:00', local_directory='/glade/scratch/$USER/temp/')
## create 30 workers (I've tried up to 70)
cluster.start_workers(30)
## attach client
client = dask.distributed.Client(cluster)
## create a rolling 30-day sum
mfds_data_30day_rolling_object = mfds_data.rolling(time=30, center=True)
mfds_data_30day_rolling_sum = client.persist(mfds_data_30day_rolling_object.sum())
, даже если я загружаю один файл (не полный набор данных), последняя строка кода, представленная выше, выглядит завершенной через панель мониторинга dask, но в конечном итоге задание будет уничтожено системой.Администраторы HPC неоднократно писали по электронной почте, говоря, что я использую слишком много памяти на узлах входа в систему, но я тщательно указал свои $TMPDIR и local_directory, и я не знаю, как это исправить.
У меня к этому есть пара вопросов:
- Есть ли лучший подход к этим вычислениям, чем те шаги, которые я предпринимаю?(Похоже, что создание ленивого объекта скользящего окна, затем с вызовом sum / persist, было бы хорошим способом. Я также пытался использовать
scatter() и futures, но проблемы с памятью сохраняются.) - Когда я смотрю на использование диском моего рабочего каталога (
du -hs /glade/scratch/$USER/temp), кажется, что рабочие не занимают какую-либо память во время sum() и persist().В этом случае, где именно является используемой памятью?У меня есть ощущение, что если я смогу это понять, я смогу решить свою проблему. - Дополнительная информация:
client.scheduler_info() подтверждает, что мой рабочий каталог был правильно установлен, и в нем созданы рабочие файлы ****, но размер папки остается фиксированным на уровне 32KB.Возможно, у меня нет представления о том, какие настройки local_directory выполняет в функции PBSCluster()?