Scale with Dask¶
The Planetary Computer Hub is a JupyterHub paired with Dask Gateway for easily creating Dask clusters to distribute your computation on a cluster of machines.
Creating a cluster¶
Use dask_gateway.GatewayCluster to quickly create a Dask cluster.
import dask_gateway
cluster = dask_gateway.GatewayCluster()
client = cluster.get_client()
cluster.scale(4)
print(cluster.dashboard_link)
GatewayCluster<staging.549920ff1ac94c1aa92f9d543147c8a4, status=running>
Don't forget the client = cluster.get_client() line. That's what actually ensures the cluster will be used for computations using Dask. Otherwise, you'll end up using Dask's local scheduler. This will run the computation on using multiple threads on a single machine, rather than the cluster. When you're using a cluster, make sure to always use the Dashboard (more below). If you aren't seeing any tasks in the dashboard, you might have forgotten to create a Dask client.
Open the dashboard¶
The Dask Dashboard provides invaluable information on the activity of your cluster. Clicking the "Dashboard" link above will open the Dask dashboard a new browser tab.
We also include the dask-labextension for laying out the Dask dashboard as tabs in the Jupyterlab workspace.
To using the dask-labextension, copy the "Dashboard" address from the cluster repr, click the orange Dask logo on the lefthand navigation bar, and paste the dashboard address
You can close your cluster, freeing up its resources, by calling cluster.close().
cluster.close()
Autoscale the cluster to your workload¶
Dask Clusters can automatically adapt the cluster size based on the size of the workload. Use cluster.adapt(minimum, maximum) to enable adaptive mode.
import dask_gateway
cluster = dask_gateway.GatewayCluster()
client = cluster.get_client()
cluster.adapt(minimum=2, maximum=50)
Dask will add workers as necessary when a computation is submitted. As an example, we'll compute the minimum daily temperature averaged over all of Hawaii, using the Daymet dataset.
import pystac_client
import planetary_computer
import xarray as xr
account_name = "daymeteuwest"
container_name = "daymet-zarr"
catalog = pystac_client.Client.open(
"https://planetarycomputer.microsoft.com/api/stac/v1",
modifier=planetary_computer.sign_inplace,
)
asset = catalog.get_collection("daymet-daily-hi").assets["zarr-abfs"]
ds = xr.open_zarr(
asset.href,
**asset.extra_fields["xarray:open_kwargs"],
storage_options=asset.extra_fields["xarray:storage_options"]
)
ds
<xarray.Dataset>
Dimensions: (nv: 2, time: 14965, x: 284, y: 584)
Coordinates:
lat (y, x) float32 dask.array<chunksize=(584, 284), meta=np.ndarray>
lon (y, x) float32 dask.array<chunksize=(584, 284), meta=np.ndarray>
* time (time) datetime64[ns] 1980-01-01T12:00:00 ... 20...
* x (x) float32 -5.802e+06 -5.801e+06 ... -5.519e+06
* y (y) float32 -3.9e+04 -4e+04 ... -6.21e+05 -6.22e+05
Dimensions without coordinates: nv
Data variables:
dayl (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
lambert_conformal_conic int16 ...
prcp (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
srad (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
swe (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
time_bnds (time, nv) datetime64[ns] dask.array<chunksize=(365, 2), meta=np.ndarray>
tmax (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
tmin (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
vp (time, y, x) float32 dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
yearday (time) int16 dask.array<chunksize=(365,), meta=np.ndarray>
Attributes:
Conventions: CF-1.6
Version_data: Daymet Data Version 4.0
Version_software: Daymet Software Version 4.0
citation: Please see http://daymet.ornl.gov/ for current Daymet ...
references: Please see http://daymet.ornl.gov/ for current informa...
source: Daymet Software Version 4.0
start_year: 1980- nv: 2
- time: 14965
- x: 284
- y: 584
- lat(y, x)float32dask.array<chunksize=(584, 284), meta=np.ndarray>
- long_name :
- latitude coordinate
- standard_name :
- latitude
- units :
- degrees_north
Array Chunk Bytes 647.88 kiB 647.88 kiB Shape (584, 284) (584, 284) Count 2 Tasks 1 Chunks Type float32 numpy.ndarray - lon(y, x)float32dask.array<chunksize=(584, 284), meta=np.ndarray>
- long_name :
- longitude coordinate
- standard_name :
- longitude
- units :
- degrees_east
Array Chunk Bytes 647.88 kiB 647.88 kiB Shape (584, 284) (584, 284) Count 2 Tasks 1 Chunks Type float32 numpy.ndarray - time(time)datetime64[ns]1980-01-01T12:00:00 ... 2020-12-...
- bounds :
- time_bnds
- long_name :
- 24-hour day based on local time
- standard_name :
- time
array(['1980-01-01T12:00:00.000000000', '1980-01-02T12:00:00.000000000', '1980-01-03T12:00:00.000000000', ..., '2020-12-28T12:00:00.000000000', '2020-12-29T12:00:00.000000000', '2020-12-30T12:00:00.000000000'], dtype='datetime64[ns]') - x(x)float32-5.802e+06 ... -5.519e+06
- long_name :
- x coordinate of projection
- standard_name :
- projection_x_coordinate
- units :
- m
array([-5802250., -5801250., -5800250., ..., -5521250., -5520250., -5519250.], dtype=float32) - y(y)float32-3.9e+04 -4e+04 ... -6.22e+05
- long_name :
- y coordinate of projection
- standard_name :
- projection_y_coordinate
- units :
- m
array([ -39000., -40000., -41000., ..., -620000., -621000., -622000.], dtype=float32)
- dayl(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean
- grid_mapping :
- lambert_conformal_conic
- long_name :
- daylength
- units :
- s
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - lambert_conformal_conic()int16...
- false_easting :
- 0.0
- false_northing :
- 0.0
- grid_mapping_name :
- lambert_conformal_conic
- inverse_flattening :
- 298.257223563
- latitude_of_projection_origin :
- 42.5
- longitude_of_central_meridian :
- -100.0
- semi_major_axis :
- 6378137.0
- standard_parallel :
- [25.0, 60.0]
array(-32767, dtype=int16)
- prcp(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean time: sum
- grid_mapping :
- lambert_conformal_conic
- long_name :
- daily total precipitation
- units :
- mm/day
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - srad(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean time: mean
- grid_mapping :
- lambert_conformal_conic
- long_name :
- daylight average incident shortwave radiation
- units :
- W/m2
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - swe(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean time: mean
- grid_mapping :
- lambert_conformal_conic
- long_name :
- snow water equivalent
- units :
- kg/m2
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - time_bnds(time, nv)datetime64[ns]dask.array<chunksize=(365, 2), meta=np.ndarray>
Array Chunk Bytes 233.83 kiB 5.70 kiB Shape (14965, 2) (365, 2) Count 42 Tasks 41 Chunks Type datetime64[ns] numpy.ndarray - tmax(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean time: maximum
- grid_mapping :
- lambert_conformal_conic
- long_name :
- daily maximum temperature
- units :
- degrees C
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - tmin(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean time: minimum
- grid_mapping :
- lambert_conformal_conic
- long_name :
- daily minimum temperature
- units :
- degrees C
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - vp(time, y, x)float32dask.array<chunksize=(365, 584, 284), meta=np.ndarray>
- cell_methods :
- area: mean time: mean
- grid_mapping :
- lambert_conformal_conic
- long_name :
- daily average vapor pressure
- units :
- Pa
Array Chunk Bytes 9.25 GiB 230.93 MiB Shape (14965, 584, 284) (365, 584, 284) Count 42 Tasks 41 Chunks Type float32 numpy.ndarray - yearday(time)int16dask.array<chunksize=(365,), meta=np.ndarray>
- long_name :
- day of year (DOY) starting with day 1 on Januaray 1st
Array Chunk Bytes 29.23 kiB 730 B Shape (14965,) (365,) Count 42 Tasks 41 Chunks Type int16 numpy.ndarray
- Conventions :
- CF-1.6
- Version_data :
- Daymet Data Version 4.0
- Version_software :
- Daymet Software Version 4.0
- citation :
- Please see http://daymet.ornl.gov/ for current Daymet data citation information
- references :
- Please see http://daymet.ornl.gov/ for current information on Daymet references
- source :
- Daymet Software Version 4.0
- start_year :
- 1980
The .compute() call in the next cell is what triggers computation and causes Dask to scale the cluster up to a dozen or so workers.
timeseries = ds["tmin"].mean(dim=["x", "y"]).compute()
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(12, 6))
timeseries.plot(ax=ax);
cluster.close()
Customize your cluster¶
dask_gateway.GatewayCluster creates a cluster with some default settings, which might not be appropriate for your workload. For example, we might have a memory-intensive workload which requires more memory per CPU core. Or we might need to set environment variables on the workers.
To customize your cluster, create a Gateway object and then customize the options.
import dask_gateway
gateway = dask_gateway.Gateway()
cluster_options = gateway.cluster_options()
In a Jupyter Notebook, you can use the HTML widget to customize the options. Or using Python you can adjust the values programmatically. We'll ask for 16GiB of memory per worker.
cluster_options["worker_memory"] = 16
Now create your cluster. Make sure to pass the cluster_options object to gateway.new_cluster.
cluster = gateway.new_cluster(cluster_options)
client = cluster.get_client()
cluster.scale(2)
Learn more¶
The Dask documentation has much more information on using Dask for scalable computing. This JupyterHub deployment uses Dask Gateway to manage creating Dask clusters.