Experiments with AWS FSx for Lustre: I/O Benchmark and Dask Cluster Deployment

1.1 AWS ParallelCluster configuration

The ~/.parallelcluster/config file is almost the same as in previous post. The only thing new is the fsx section. At the time of writing, only CentOS 7 is supported 5.

[cluster your-cluster-section-name]
base_os = centos7
...
fsx_settings = fs

[fsx fs]
shared_dir = /fsx
storage_capacity = 14400

According to AWS 6, a 14,400 GB file system would deliver 2,880 MB/s throughput. You may use a much bigger size for production, but this size should be good for initial testing.

The above config will create an empty file system. A more interesting usage is to pre-import S3 buckets:

[fsx fs]
shared_dir = /fsx
storage_capacity = 14400
imported_file_chunk_size = 1024
import_path = s3://era5-pds

Here I am using the ECMWF ERA5 Reanalysis data 7. The data format is NetCDF4. Other datasets would work similarly.

pcluster create can take several minutes longer than without FSx, because provisioning a Lustre server probably involves heavy-lifting on the AWS side. Go to the "Amazon FSx" console to check the creation status.

After a successful launch, log in and run df -h to make sure that Lustre is properly mounted to /fsx.

1.2 Testing integration with existing S3 bucket

Objects in the bucket will appear as normal on-disk files.

$ cd /fsx
$ ls # displays objects in the S3 bucket
...

However, FSx only stores metadata:

$ cd /fsx/2008/01/data # specific to ERA5 data
$ ls -lh *   # the data appear to be big (~1 GB)
-rwxr-xr-x 1 root root  988M Jul  4  2018 air_pressure_at_mean_sea_level.nc
...
$ du -sh *  # but the actual content is super small.
512 air_pressure_at_mean_sea_level.nc
...

The actual data will be pulled from S3 when accessed. For a NetCDF4 file, either ncdump -h or h5ls will display its basic contents and cause the entire file to be pulled from S3.

$ ncdump -h air_pressure_at_mean_sea_level.nc  # `ncdump` is installable from `sudo yum install netcdf`, or from Spack, or from Conda
...
$ du -sh *  # now much bigger
962M        air_pressure_at_mean_sea_level.nc
...

2.1 Install IOR by Spack

Configure Spack as in the previous post. Then, getting IOR is simply:

$ spack install ior ^openmpi+pmi schedulers=slurm

IOR is also quite easy to install from source, outside of Spack.

Discover the ior executable by:

$ export PATH=$(spack location -i ior)/bin:$PATH

2.2 I/O benchmark result

With two c5n.18xlarge compute nodes running, a multi-node, parallel write-read test can be done by:

$ mkdir /fsx/ior_tempdir
$ cd /fsx/ior_tempdir
$ srun -N 2 --ntasks-per-node 36 ior -t 1m -b 16m -s 4 -F -C -e
...
Max Write: 1632.01 MiB/sec (1711.28 MB/sec)
Max Read:  1654.59 MiB/sec (1734.96 MB/sec)
...

Conducting a proper I/O benchmark is not straightforward, due to various caching effects. IOR implements several tricks (reflected in command line parameters) to get around those effects 9.

I can get maximum throughput with 8 client nodes:

$ srun -N 8 --ntasks-per-node 36 ior -t 1m -b 16m -s 4 -F -C -e
...
Max Write: 2905.59 MiB/sec (3046.73 MB/sec)
Max Read:  2879.96 MiB/sec (3019.85 MB/sec)
...

This matches the 2,880 MB/s claimed by AWS! Using more nodes shows marginal improvement, since the bandwidth should already be saturated.

The logical next step is to test IO-heavy HPC applications and conduct a detailed I/O-profiling. In this post, however, I decide to try a more interesting use case -- big data analytics.

3.1 Cluster deployment with Dask-jobqueue

The deployment turns out to be extremely easy. I am still in the learning curve of Kubernetes, and this alternative HPC approach feels much more straightforward for an HPC person like me.

First, get Miniconda:

$ cd /shared
$ wget https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh -O miniconda.sh
$ bash miniconda.sh -b -p miniconda
$ echo ". /shared/miniconda/etc/profile.d/conda.sh" >> ~/.bashrc
$ source ~/.bashrc
$ conda create -n py37 python=3.7
$ conda activate py37  # replaces `source activate` for conda>=4.4
$ conda install -c conda-forge xarray netCDF4 cartopy dask-jobqueue jupyter

Optionally, install additional visualization libraries that I will use later:

$ pip install geoviews hvplot datashader

With two idle c5n.18xlarge nodes, use the following code in ipython to initialize a distributed cluster:

from dask_jobqueue import SLURMCluster
cluster = SLURMCluster(cores=72, processes=36, memory='150GB')  # Slurm thinks there are 72 cores per node due to EC2 hyperthreading
cluster.scale(2*36)

from distributed import Client
client = Client(cluster)

In a separate shell, use sinfo to check the node status -- they should be fully allocated.

To enable Dask's dashboard 12, add an additional SSH connection in a new shell:

$ pcluster ssh your-cluster-name -N -L 8787:localhost:8787

Visit localhost:8787 in the web browser (NOT something like http://172.31.5.224:8787 shown in Python) .

Alternatively, everything can be put together, including Jupyter notebook's port-forwarding:

$ pcluster ssh your-cluster-name -L 8889:localhost:8889 -L 8787:localhost:8787
$ conda activate py37
$ jupyter notebook --NotebookApp.token='' --no-browser --port=8889

Visit localhost:8889 to use the notebook.

That's all about the deployment! This Dask cluster is able to perform parallel read/write with the Lustre file system.

3.2 Computation with Dask-distributed

As an example, I compute the average Sea Surface Temperature (SST) 13 over near 300 GBs of ERA5 data. It gets done in 15 seconds with 8 compute nodes, which would have taken > 20 minutes with a single small node. Here's the screen recording of Dask dashboard during computation.

The full code is available in the next notebook, with some technical comments. At the end of the notebook also shows a sign of climate change (computed from the SST data), so at least we get a bit scientific insight from this toy problem. Hopefully such great computing power can be used to solve some big science.