diff --git a/docs/cache_hash.rst b/docs/cache_hash.rst deleted file mode 100644 index ad86737..0000000 --- a/docs/cache_hash.rst +++ /dev/null @@ -1,87 +0,0 @@ -How to add custom hash functions for cacheable -============================================== - -The cacheable decorator uses a hash function to generate a key for each of -the arguments passed to the function along with the return value. This hash -function is internally decided based on the type of the argument. However, this -decision can be a bit unreliable if the arguments or return value have custom -or complex data types. The data types which can be recognized automatically -(so you don't need to pass any classes). Before we move ahead, let's see how -the cacheable decorator works. - -.. code-block:: python - - from scalable import * - - - # No arguments are passed to the decorator because the arguments in the function - # are of primitive data types which are recognized by the decorator. - @cacheable - def add(a: int, b: int): - return a + b - - - # Since the argument of the check function is a custom data type, it would be - # best to pass the person class to the decorator so that the person object - # passed to the function can be hashed properly. It is assumed that the Person - # class has a __hash__() method. - @cacheable(person=Person) - def check(person: Person): - if person.name is None: - return False - return True - - # The decorator is passed all the argument types and the return value data type. - # The argument types are not needed in this specific case because the arguments - # are of primitive data types which are recognized by the decorator. - @cacheable(name=str, age=int, return_type=Person) - def make_person(name: str, age: int): - return Person(name, age) - - - -The `add` function is a simple function that takes two integers and returns -their sum. int is a primitive data type and is recognized by the decorator. -The `check` function takes a custom data type `Person` as an argument. The -`Person` class is passed to the decorator as the value for the `person` -keyword which matches the argument name in the function. - -The `Person` class should have an implemented `__hash__` method that ideally -returns a unique hash for each object. As an example, let's make an example -Person class and implement a `__hash__` method for it. - -.. code-block:: python - - # xxhash is used here but any hashing library can be used (hashlib is built-in) - # However, xxhash is a non-cryptographic hash library which is much faster - # than hashlib or other cryptographic libraries. - # xxhash can be installed by running `pip install xxhash` - - from xxhash import xxh32 - import scalable - - class Person: - def __init__(self, name: str, age: int): - self.name = name - self.age = age - - def __hash__(self): - digest = 0 - # It is important to use the same seed value at each invocation. An - # easy way to ensure this is to use the seed value scalable. - # However, any constant value for seed can be used. Some value must be - # used to ensure that the hash is consistent across different - # invocations. - x = xxh32(seed=scalable.SEED) - x.update(self.name.encode('utf-8')) - x.update(str(self.age).encode('utf-8')) - digest = x.intdigest() - return digest - -The `Person` class with the `__hash__` method can be directly passed to the -decorator. This would allow the decorator to hash the `Person` object properly. - -For any data types that are used in arguments or return values, the data type -class should be passed to the decorator for the most reliable behavior. If -needed, please feel free to open an issue -`here `_. diff --git a/docs/container.rst b/docs/container.rst deleted file mode 100644 index 87fd34f..0000000 --- a/docs/container.rst +++ /dev/null @@ -1,88 +0,0 @@ -How to add custom containers for scalable as workers -==================================================== - -Containers are quite central to scalable. Everything runs in a container to -maintain isolation and reproducibility. A Dockerfile is provided with scalable -which has targets for in-house JGCRI energy models. However, these targets can -either be modified or extended to have custom ones. In the provided Dockerfile, -all the targets are built from a base or "build_env" target. Let's look at this -target: - -.. code-block:: dockerfile - - FROM ubuntu:22.04 AS build_env - - ENV DEBIAN_FRONTEND=noninteractive - RUN apt-get -y update && apt-get install -y \ - wget unzip openjdk-11-jdk-headless build-essential libtbb-dev \ - libboost-dev libboost-filesystem-dev libboost-system-dev \ - python3 python3-pip libboost-python-dev libboost-numpy-dev \ - ssh nano locate curl net-tools netcat-traditional git python3 \ - python3-pip python3-dev gcc libboost-python1.74 libboost-numpy1.74 \ - openjdk-11-jre-headless libtbb12 rsync - RUN apt-get -y update && apt -y upgrade - RUN pip3 install --upgrade dask[complete] dask-jobqueue dask_mpi pyyaml \ - joblib versioneer tomli xarray - RUN apt-get -y update && apt -y upgrade - RUN apt-get install -y --no-install-recommends r-base r-base-dev - RUN apt-get install -y --no-install-recommends python3-rpy2 - ENV R_LIBS_USER usr/lib/R/site-library - RUN chmod a+w /usr/lib/R/site-library - RUN cp -r /usr/lib/R/site-library /usr/local/lib/R/site-library - RUN python3 -m pip install -U pip - - -The target above details the libraries and packages needed for most JGCRI -energy models. Not all of these libraries are needed for any one model but -most are useful in general. Now, let's look at a target for scalable itself and -a specific model, in this case demeter: - -.. code-block:: dockerfile - - FROM build_env AS scalable - ADD "https://api.github.com/repos/JGCRI/scalable/commits?per_page=1" latest_commit - RUN git clone https://github.com/JGCRI/scalable.git /scalable - RUN pip3 install /scalable/. - RUN pip3 install --force-reinstall xarray==2024.5.0 - RUN pip3 install --force-reinstall numpy==1.26.4 - - FROM build_env AS demeter - RUN apt-get -y update && apt -y upgrade - RUN python3 -m pip install git+http://github.com/JGCRI/demeter.git#egg=demeter - RUN mkdir /demeter - RUN echo "import demeter" >> /demeter/install_script.py \ - && echo "demeter.get_package_data(\"/demeter\")" >> /demeter/install_script.py - RUN python3 /demeter/install_script.py - COPY --from=scalable /scalable /scalable - RUN pip3 install /scalable/. - RUN pip3 install --force-reinstall xarray==2024.5.0 - RUN pip3 install --force-reinstall numpy==1.26.4 - -Right off the bat, one thing that stands out is the force reinstallation of -xarray and numpy. This is because the versions of these libraries used in most, -if not all, JGCRI energy models are not the latest. So, this is done to ensure -that nothing breaks when the models are ran. While different containers can -have different environments, libraries like numpy, xarry, and dask are used -by all models and should have the same versions in all the containers. Python -versions should also be the same in all the containers. - -The scalable target simply installs the latest version of the scalable package -and corrects for any library versions at the end. Please feel free to remove -the force reinstallation of numpy and xarray if you are sure that the models -you are running will not break with the latest versions of these libraries. - -The demeter target also just installs the latest version of the demeter along -with installing its package data. The scalable package is then copied from -the scalable target to the demeter target. This is important and it's highly -recommended to do the same in any custom target. Along with ensuring that the -scalable package is installed in the target container which is needed, it also -ensures that the same version of the scalable package is used in all the -targets. - -For a custom target, a similar template as demeter should be followed. Start -with the build_env target, add the desired libraries or applications, and copy -the scalable package from the scalable target. Ensure to have a scalable target -which is named "scalable". It's used by the bootstrap script. If any library -version corrections are needed, they can be made at the end of the target. -The reinstallation commands should be at the end to prevent any unintended -side effects by other commands. diff --git a/docs/demo.rst b/docs/demo.rst deleted file mode 100644 index e52c12a..0000000 --- a/docs/demo.rst +++ /dev/null @@ -1,402 +0,0 @@ -Scalable Example Workflow -========================= - -This demo is presented as a sequence of small steps so you can follow and run -each stage independently. - -The workflow demonstrates how Scalable coordinates a small integrated modeling -pipeline across multiple software environments. GCAM runs first to generate a -model database, a lightweight extraction step reads the needed time series from -that database, and Stitches uses the extracted trajectory to build gridded -weather outputs. Each stage is submitted as a normal Python function, but -Scalable decides which container profile should execute it and passes results -between stages as Dask futures. - -Step 1: Imports and logging ---------------------------- - -The first step imports Scalable's public API and a few modules used by the -example functions. ``from scalable import *`` provides the cluster, client, -caching decorators, and type helpers used later in the workflow. ``os`` is used -inside worker functions for path handling, and ``gcam_config`` represents a -project-specific module that describes how GCAM configuration files should be -hashed for caching. - -Logging is set to ``DEBUG`` so the example prints more information about worker -startup, task submission, container selection, and any errors that occur. During -development this is useful because HPC failures can otherwise appear as delayed -or silent worker timeouts. - -.. code-block:: python - - from scalable import * - - import logging - import os - import gcam_config - - logging.basicConfig(level=logging.DEBUG) - - -Step 2: Create the cluster --------------------------- - -The ``SlurmCluster`` object describes how Scalable should request resources from -the HPC scheduler. The ``queue``, ``walltime``, and ``account`` values map to the -same concepts you would normally provide in a Slurm batch script. This object -does not yet start model work; it records the scheduler settings that will be -used later when workers are added. - -``silence_logs=False`` keeps worker and scheduler output visible. That is -helpful for a tutorial because it exposes what Scalable is doing behind the -scenes, including whether workers were accepted by Slurm and whether they -connected back to the Dask scheduler. - -.. code-block:: python - - cluster = SlurmCluster( - queue='short', - walltime='02:00:00', - account='GCIMS', - silence_logs=False, - ) - - -Step 3: Register container profiles ------------------------------------ - -Use one container profile per software environment. A container profile tells -Scalable which image target to use, how many CPUs and how much memory each -worker should reserve, and which host directories should be mounted inside the -container. Tags such as ``"gcam"`` and ``"stitches"`` become routing labels: when -a task is submitted with a matching tag, Scalable sends it to workers running -that environment. - -Directory mappings are written as ``host_path: container_path``. Functions that -run inside the container should use the container paths, not the original host -paths. This keeps code consistent even if the host filesystem layout differs -between local setup and the HPC system. - -.. code-block:: python - - # GCAM can use multiple threads, so this profile reserves 6 CPUs. - cluster.add_container( - tag="gcam", - cpus=6, - memory="20G", - dirs={ - "/path/to/gcam-core/exe": "/gcam-core/exe", - "/path/to/gcam-core/input": "/gcam-core/input", - "/path/to/gcam-core/output": "/gcam-core/output", - "/path/to/shared/data": "/data", - }, - ) - -.. code-block:: python - - # Stitches is typically single-threaded in this workflow. - cluster.add_container( - tag="stitches", - cpus=1, - memory="50G", - dirs={"/path/to/shared/data": "/data", "/path/to/archive/data": "/archive"}, - ) - -.. code-block:: python - - # Xanthos profile (not used below, shown as an additional profile example). - cluster.add_container( - tag="xanthos", - cpus=1, - memory="20G", - dirs={ - "/path/to/shared/data": "/data", - "/path/to/archive/data": "/archive", - "/path/to/project/scratch": "/scratch", - }, - ) - - -Step 4: Define worker functions -------------------------------- - -Define each function once, then submit them in dependency order. These functions -are normal Python functions, but they should import large or container-specific -dependencies inside the function body. Doing so ensures imports happen on the -worker inside the correct container rather than on the login node or client -process. - -The ``@cacheable`` decorator marks expensive deterministic steps whose outputs -can be reused. When Scalable sees the same function inputs and compatible type -metadata, it can avoid repeating work and return the cached output instead. The -``return_type`` and custom type hints, such as ``config_file=gcam_config.GcamConfig``, -help Scalable hash non-trivial inputs and outputs reliably. - -The first function runs GCAM for a requested model period and returns the path to -the generated database directory. ``get_worker().id`` is used in the database -name so concurrent GCAM workers do not overwrite one another's output. - -.. code-block:: python - - @cacheable(return_type=DirType, config_file=gcam_config.GcamConfig) - def run_gcam(config_file, period): - import gcamwrapper as gw - from dask.distributed import get_worker - - g = gw.Gcam(os.path.basename(config_file), "/gcam-core/exe") - g.run_period(g.convert_year_to_period(period)) - dbname = "/gcam-core/output/" + get_worker().id + "database" - return g.print_xmldb(dbname) - -The database reader runs after GCAM completes. It receives the database path -returned by ``run_gcam``, opens it with ``gcamreader``, selects one query from a -batch-query XML file, and returns the query result as tabular data. Because the -input may be a future, Scalable waits for the upstream GCAM task before running -this function. - -.. code-block:: python - - def readdb(db_path): - import gcamreader - import os - - conn = gcamreader.LocalDBConn(os.path.dirname(db_path), os.path.basename(db_path)) - query = gcamreader.parse_batch_query("/path/to/sample-queries.xml")[2] - return conn.runQuery(query) - -The interpolation helper converts sparse model-year output into an annual time -series. Stitches expects a continuous trajectory, so this helper fills the years -between GCAM time points with linearly interpolated values. - -.. code-block:: python - - def interp(years, values): - import numpy as np - - min_year = min(years) - max_year = max(years) - new_years = np.arange(min_year, max_year + 1) - new_values = np.zeros(len(new_years)) - - for index, year in enumerate(new_years): - if np.isin(year, years): - new_values[index] = values[year == years][0] - else: - less_year = max(years[year > years]) - more_year = min(years[year < years]) - less_value = values[np.where(less_year == years)[0][0]] - more_value = values[np.where(more_year == years)[0][0]] - p = (year - less_year) / (more_year - less_year) - new_values[index] = p * more_value + (1 - p) * less_value - - return new_years, new_values - -``stitch_prep`` transforms the GCAM query result into the target format expected -by Stitches. It loads a matching archive, selects candidate weather-model data, -interpolates the GCAM trajectory, normalizes the values relative to the -1975--2014 baseline, and builds a Stitches recipe describing which archive -segments should be combined to match the target trajectory. - -This function is also cacheable because recipe generation can be expensive and -is deterministic for the same input data and archive. If the GCAM output has not -changed, rerunning the workflow can skip this preparation step. - -.. code-block:: python - - @cacheable - def stitch_prep(df): - import numpy as np - import pandas as pd - import pkg_resources - import stitches - - path = pkg_resources.resource_filename('stitches', 'data/matching_archive_staggered.csv') - data = pd.read_csv(path) - end_yr_vector = np.arange(2100, 1800, -9) - data = stitches.fx_processing.subset_archive( - staggered_archive=data, - end_yr_vector=end_yr_vector, - ) - model_data = data[(data["model"] == "CanESM5") & (data["experiment"].str.contains('ssp585'))] - - years, values = interp(np.array(df["Year"]), np.array(df["value"])) - df = pd.DataFrame({"year": years, "value": values}) - df['variable'] = 'tas' - df['model'] = '' - df['ensemble'] = '' - df['experiment'] = 'GCAM7-Ref' - df['unit'] = 'degC change from avg over 1975~2014' - df.value = df.value - np.mean(df.value[(df.year <= 2014) & (df.year >= 1975)]) - df = df[['variable', 'experiment', 'ensemble', 'model', 'year', 'value', 'unit']] - - target_chunk = stitches.fx_processing.chunk_ts(df, n=9) - target_data = stitches.fx_processing.get_chunk_info(target_chunk) - - stitches_recipe = None - for _ in range(10): - stitches_recipe = stitches.make_recipe( - target_data, - model_data, - tol=0.0, - N_matches=1, - res='day', - non_tas_variables=['tasmin', 'pr', 'hurs', 'sfcWind', 'rsds', 'rlds'], - ) - - last_period_length = ( - stitches_recipe['target_end_yr'].values[-1] - stitches_recipe['target_start_yr'].values[-1] - ) - asy = stitches_recipe['archive_start_yr'].values - asy[-1] = stitches_recipe['archive_end_yr'].values[-1] - last_period_length - stitches_recipe['archive_start_yr'] = asy.copy() - return stitches_recipe - -``run_stitches`` consumes the recipe and writes gridded stitched outputs to the -requested output directory. The function sets Dask's internal scheduler to -``"synchronous"`` while calling Stitches because Stitches may create its own Dask -work internally. Keeping that work local to the selected worker avoids common -pickling and cross-environment issues when the broader Scalable cluster contains -workers with different containers. - -.. code-block:: python - - @cacheable - def run_stitches(recipe, output_path): - import dask - import stitches - - # Synchronous mode avoids common pickling issues in mixed environments. - with dask.config.set(scheduler="synchronous"): - return stitches.gridded_stitching(output_path, recipe) - - -Step 5: Start initial workers and client ----------------------------------------- - -This step starts only the workers needed for the first stage of the workflow: -two GCAM workers. Starting a small number of workers up front keeps the resource -request focused on the immediate work instead of reserving every possible -container at once. - -The ``ScalableClient`` connects to the cluster and is used for all subsequent -task submission, dependency tracking, and result retrieval. It behaves like a -Dask client with additional Scalable routing options such as ``tag`` and ``n``. - -.. code-block:: python - - cluster.add_workers(n=2, tag="gcam") - sc_client = ScalableClient(cluster) - - -Step 6: Submit GCAM and database extraction tasks -------------------------------------------------- - -The first submitted task runs GCAM in the ``"gcam"`` container. The ``n=1`` value -means the task needs one worker slot, while ``tag="gcam"`` restricts execution -to workers created from the GCAM profile. The call returns immediately with a -future instead of blocking until GCAM finishes. - -The second task uses ``future1`` as its input. Passing futures directly is how -the workflow graph is constructed: Scalable knows ``readdb`` depends on -``run_gcam`` and will not execute the database query until GCAM has produced its -database path. This allows you to describe the pipeline without manually polling -for completion. - -.. code-block:: python - - future1 = sc_client.submit( - run_gcam, - "/path/to/gcam-core/exe/configuration_ref.xml", - 2100, - n=1, - tag="gcam", - ) - -.. code-block:: python - - future2 = sc_client.submit(readdb, future1, n=1, tag="gcam") - - -Step 7: Add stitches worker and prepare stitching inputs --------------------------------------------------------- - -After the GCAM extraction stage is in the graph, the workflow adds a worker for -the ``"stitches"`` profile. This demonstrates dynamic scaling: the Stitches -environment is started only when the workflow is about to need it. - -``stitch_prep`` is submitted to the Stitches worker and depends on ``future2``. -Once the database extraction returns tabular data, this task converts that data -into a Stitches recipe. The task runs in the Stitches container because the -function imports and calls the Stitches package. - -.. code-block:: python - - cluster.add_workers(n=1, tag="stitches") - future3 = sc_client.submit(stitch_prep, future2, n=1, tag="stitches") - - -Step 8: Scale down unused GCAM workers --------------------------------------- - -Only remove workers after downstream tasks no longer depend on them. At this -point, the GCAM run and database extraction have already been submitted, and the -remaining work is handled by the Stitches profile. Removing the GCAM workers -frees scheduler resources and reduces the chance of idle workers occupying an -allocation that another job could use. - -In a production workflow, remove workers only after you are confident no queued -or future downstream work needs that container. If later tasks still need GCAM -libraries or files that are only available in the GCAM environment, keep those -workers alive until those tasks are submitted and completed. - -.. code-block:: python - - cluster.remove_workers(n=2, tag="gcam") - - -Step 9: Run stitches and fetch final output -------------------------------------------- - -The final computational task passes the Stitches recipe future into -``run_stitches``. As with earlier steps, using a future as an argument creates a -dependency edge: Stitches will not start until recipe preparation finishes. The -output path is a container-visible directory where Stitches should write final -products. - -``future4.result()`` blocks the client process until the final task completes, -then returns the value produced by ``run_stitches``. In small examples printing -the result is convenient; in a larger workflow you might gather multiple futures, -write a manifest, or submit additional post-processing tasks instead. - -.. code-block:: python - - future4 = sc_client.submit( - run_stitches, - future3, - '/path/to/output/directory/', - n=1, - tag='stitches', - ) - -.. code-block:: python - - print(future4.result()) - - -Step 10: Close the cluster --------------------------- - -Closing the cluster shuts down the Dask client, scheduler, and any remaining -workers. This cleanup step is important on shared HPC systems because it releases -Slurm allocations and prevents orphaned worker jobs from continuing to consume -resources after the workflow has finished. - -.. code-block:: python - - cluster.close() - - -This demo shows a complete multi-stage workflow where each stage runs in the -appropriate container profile and passes futures to downstream steps. If you run -into issues, open one at -`https://github.com/JGCRI/scalable/issues `_. diff --git a/docs/helps_demo.rst b/docs/helps_demo.rst deleted file mode 100644 index dedcd96..0000000 --- a/docs/helps_demo.rst +++ /dev/null @@ -1,260 +0,0 @@ -A Complete Process of Adding & Scaling a New Application -========================================================= - -This demo is broken into clear stages for adding an R-based application -(`HELPS `_) and running it at scale with -Scalable. - -Step 1: Add a new container target ----------------------------------- - -Add a ``helps`` target to your Dockerfile (the one used by -``scalable_bootstrap``). - -.. code-block:: dockerfile - - FROM ubuntu:22.04 AS build_env - - ENV DEBIAN_FRONTEND=noninteractive - RUN apt-get -y update && apt-get install -y \ - wget unzip openjdk-11-jdk-headless build-essential libtbb-dev \ - libboost-dev libboost-filesystem-dev libboost-system-dev \ - python3 python3-pip libboost-python-dev libboost-numpy-dev \ - ssh nano locate curl net-tools netcat-traditional git python3 python3-pip \ - python3-dev gcc libboost-python1.74 libboost-numpy1.74 openjdk-11-jre-headless libtbb12 rsync - RUN apt-get -y update && apt -y upgrade - RUN pip3 install --upgrade dask[complete] dask-jobqueue dask_mpi pyyaml joblib versioneer tomli xarray - RUN apt-get -y update && apt -y upgrade - RUN mkdir -p /usr/lib/R/site-library - ENV R_LIBS_USER /usr/lib/R/site-library - RUN chmod a+w /usr/lib/R/site-library - RUN apt-get install -y --no-install-recommends r-base r-base-dev - RUN apt-get install -y --no-install-recommends python3-rpy2 - RUN python3 -m pip install -U pip - -.. code-block:: dockerfile - - FROM build_env AS helps - RUN apt-get -y update && apt -y upgrade && apt-get -y -f install - RUN apt install --fix-missing -y software-properties-common - RUN add-apt-repository ppa:ubuntugis/ppa - RUN apt-get update - ENV R_LIBS_USER /usr/lib/R/site-library - RUN chmod a+w /usr/lib/R/site-library - RUN apt-get install -y libcurl4-openssl-dev libssl-dev libxml2-dev libudunits2-dev libproj-dev libavfilter-dev \ - libbz2-dev liblzma-dev libfontconfig1-dev libharfbuzz-dev libfribidi-dev libgeos-dev libgdal-dev - RUN git clone https://github.com/JGCRI/HELPS.git /HELPS - - RUN echo "import rpy2.robjects.packages as rpackages" >> /install_script.py \ - && echo "utils = rpackages.importr('utils')" >> /install_script.py \ - && echo "utils.install_packages('devtools')" >> /install_script.py \ - && echo "utils.install_packages('arrow')" >> /install_script.py \ - && echo "utils.install_packages('assertthat')" >> /install_script.py - RUN python3 /install_script.py - - RUN echo "import rpy2.robjects.packages as rpackages" > /install_script.py \ - && echo "devtools = rpackages.importr('devtools')" >> /install_script.py \ - && echo "devtools.install_github('JGCRI/HELPS', dependencies=True)" >> /install_script.py - RUN python3 /install_script.py - - RUN echo "import rpy2.robjects.packages as rpackages" > /install_script.py \ - && echo "helps = rpackages.importr('HELPS')" >> /install_script.py - RUN python3 /install_script.py - - COPY --from=scalable /scalable /scalable - RUN pip3 install /scalable/. - RUN pip3 install --force-reinstall xarray==2024.5.0 - RUN pip3 install --force-reinstall numpy==1.26.4 - -Notes: - -* ``build_env`` is a recommended base for new targets. -* ``rpy2`` is used here for consistency with runtime access patterns. -* Keep core package version pins aligned with other Scalable targets. - - -Step 2: Build and publish the new target ----------------------------------------- - -Run ``scalable_bootstrap`` and select ``helps`` when prompted for targets. -Bootstrap will build locally and upload to the remote system. - - -Step 3: Create a preload script for R imports ---------------------------------------------- - -Preloading R objects avoids repeated session initialization and reduces runtime -issues. - -.. code-block:: python - - def dask_setup(worker): - import rpy2.robjects.packages as rpackages - import rpy2.robjects as r - - helps = rpackages.importr('HELPS') - worker.imports = {} - worker.imports['HELPS'] = helps - worker.imports['country_raster'] = r.r['country_raster'] - worker.imports['reg_WB_raster'] = r.r['reg_WB_raster'] - - -Step 4: Define task functions ------------------------------ - -Each function pulls R objects from ``worker.imports``. - -.. code-block:: python - - from scalable import * - - def run_heatstress(hurs_file_name, tas_file_name, rsds_file_name, target_year): - worker = get_worker() - helps = worker.imports['HELPS'] - - return helps.cal_heat_stress( - TempRes="month", - SECTOR="SUNF_R", - HS=helps.WBGT_ESI, - YEAR_INPUT=target_year, - a=hurs_file_name, - b=tas_file_name, - c=rsds_file_name, - ) - -.. code-block:: python - - def run_physical_work_capacity(heat_stress_raster): - worker = get_worker() - helps = worker.imports['HELPS'] - - return helps.cal_pwc( - WBGT=heat_stress_raster, - LHR=helps.LHR_Hothaps, - workload="high", - ) - -.. code-block:: python - - def run_annualized_physical_work_capacity(physical_work_capacity_raster): - worker = get_worker() - helps = worker.imports['HELPS'] - - return helps.monthly_to_annual( - input_rack=physical_work_capacity_raster, - SECTOR="SUNF_R", - ) - -.. code-block:: python - - def run_country_physical_work_capacity(annualized_physical_work_capacity_df): - worker = get_worker() - helps = worker.imports['HELPS'] - country_raster = worker.imports['country_raster'] - - return helps.grid_to_region( - grid_annual_value=annualized_physical_work_capacity_df, - SECTOR="SUNF_R", - rast_boundary=country_raster, - ) - -.. code-block:: python - - def run_basin_physical_work_capacity(annualized_physical_work_capacity_df): - worker = get_worker() - helps = worker.imports['HELPS'] - reg_WB_raster = worker.imports['reg_WB_raster'] - - return helps.grid_to_region( - grid_annual_value=annualized_physical_work_capacity_df, - SECTOR="SUNF_R", - rast_boundary=reg_WB_raster, - ) - - -Step 5: Configure runtime inputs --------------------------------- - -.. code-block:: python - - hurs_file_name = "path/to/hurs_file" - tas_file_name = "path/to/tas_file" - rsds_file_name = "path/to/rsds_file" - - # 2015 through 2100 in 5-year increments - target_years = list(range(2015, 2105, 5)) - num_years = len(target_years) - - -Step 6: Create cluster and add HELPS container profile -------------------------------------------------------- - -.. code-block:: python - - cluster = SlurmCluster( - queue='short', - walltime='02:00:00', - account='GCIMS', - silence_logs=False, - ) - - cluster.add_container( - tag="helps", - cpus=1, # R workloads are commonly single-threaded - memory="8G", - preload_script='/path/to/preload_script.py', - dirs={"/path1": "/path1", "/path2": "/path2"}, - ) - - -Step 7: Start workers and create a client ------------------------------------------ - -.. code-block:: python - - cluster.add_workers(n=num_years, tag="helps") - sc_client = ScalableClient(cluster) - - -Step 8: Submit the pipeline using ``map`` ------------------------------------------ - -Use ``map`` when running the same function over a sequence of arguments. - -.. code-block:: python - - heatstress_futures = sc_client.map( - run_heatstress, - [hurs_file_name] * num_years, - [tas_file_name] * num_years, - [rsds_file_name] * num_years, - target_years, - n=1, - tag="helps", - ) - -.. code-block:: python - - pwc_futures = sc_client.map(run_physical_work_capacity, heatstress_futures, n=1, tag="helps") - annualized_pwc_futures = sc_client.map(run_annualized_physical_work_capacity, pwc_futures, n=1, tag="helps") - country_pwc_futures = sc_client.map(run_country_physical_work_capacity, annualized_pwc_futures, n=1, tag="helps") - basin_pwc_futures = sc_client.map(run_basin_physical_work_capacity, annualized_pwc_futures, n=1, tag="helps") - - -Step 9: Gather results ----------------------- - -.. code-block:: python - - heatstress_results = sc_client.gather(heatstress_futures) - pwc_results = sc_client.gather(pwc_futures) - annualized_pwc_results = sc_client.gather(annualized_pwc_futures) - country_pwc_results = sc_client.gather(country_pwc_futures) - basin_pwc_results = sc_client.gather(basin_pwc_futures) - - -This workflow demonstrates how to add a new R-based container, preload R -dependencies, and scale a year-by-year analysis pipeline through Scalable. -For troubleshooting, open an issue at -`https://github.com/JGCRI/scalable/issues `_. - diff --git a/docs/index.rst b/docs/index.rst index 35630f6..74cf13b 100755 --- a/docs/index.rst +++ b/docs/index.rst @@ -72,30 +72,3 @@ Contents caching functions -.. _how_tos_section: - -.. toctree:: - :caption: How-tos - :maxdepth: 1 - - cache_hash - container - rpy2 - -.. _demos_section: - -.. toctree:: - :caption: Demos - :maxdepth: 1 - - demo - helps_demo - -.. _common_issues_section: - -.. toctree:: - :caption: Common Issues - :maxdepth: 1 - - issues - diff --git a/docs/issues.rst b/docs/issues.rst deleted file mode 100644 index 4886693..0000000 --- a/docs/issues.rst +++ /dev/null @@ -1,91 +0,0 @@ -Common Issues -============= - -Pickling Error --------------- - -This page outlines some of the common problems and caveats still present in -the current version of scalable. While some of them are being worked on, others -may be inherent to dask. - -To start with, let's look at something which we is already used in the -:doc:`demo`: - -.. code-block:: python - - @cacheable - def run_stitches(recipe, output_path): - import stitches - import dask - ## The dask config is set to synchronous to avoid any issues. - with dask.config.set(scheduler="synchronous"): - outputs = stitches.gridded_stitching(output_path, recipe) - return outputs - -The above code is a simple function that runs -`stitches `_. The primary code line which -runs sitches is ran under the dask.config.set context manager. The scheduler is -set to synchronous in this case. The alternative would've been to write this -function as: - -.. code-block:: python - - @cacheable - def run_stitches(recipe, output_path): - import stitches - outputs = stitches.gridded_stitching(output_path, recipe) - return outputs - -The above code should've worked well. However, the following error is thrown -when the function is called with a dask client (scalable): - -.. image:: images/error1.png - :align: center - -The error thrown above is a pickling error. This happens because dask tries to -use multiple different workers to make the dask task graph. However, since our -workers have different environments, the `run_stitches` task cannot be pickled -by other workers. Therefore, whenever this issue is encountered, it is -recommended to set the scheduler to be "synchronous" which means that it will -pickle the task and run it on the same specified worker. - -Deadlocks and Stuck Workers ---------------------------- - -Another issue that can occasionally be encountered is deadlocks. It is possible -that a worker can get stuck processing a function forever. This issue is quite -subtle to debug as the worker can just genuinely be processing instead of being -stuck. The best way to recognize this issue is if the function is taking a lot -longer than expected. The most common cause of this issue is a deadlock. And -the most common cause of a deadlock is assigning multiple CPUs to a container -which is only running functions which use a single CPU. In other words, -reserving multiple CPUs for single-threaded functions can cause deadlocks. - -When multiple CPUs are assigned to a container in the following way: - -.. code-block:: python - - cluster.add_container( - tag="container_tag", - cpus=2, - memory="50G", - dirs={"/path/to/shared/data": "/data", "/path/to/archive/data": "/archive"}, - ) - -The scalable backend assigns a threadpool with 2 threads to any worker with the -same tag as the container. However, if the functions running on the worker are -single-threaded, there could be two instances of the function assigned to the -worker which can occasionally cause a deadlock. To prevent this, please ensure -that the correct amount of CPUs are assigned to the container. - -General Errors --------------- - -There can also be just general errors which are either thrown by dask or are -manifested in the form of workers which didn't connect or slurm errors. There -are mechanisms within Scalable which should warn about any workers which -couldn't connect for whatever reason. However, as a rule of thumb, restarting -the cluster and the workflow is the best way to resolve any one time errors. -HPC systems can be unreliable and throw unknown errors sometimes. As always, -please feel free to open an issue -`here `_ for any persistent issues. diff --git a/docs/rpy2.rst b/docs/rpy2.rst deleted file mode 100644 index 8a06f3d..0000000 --- a/docs/rpy2.rst +++ /dev/null @@ -1,232 +0,0 @@ -How to convert R code to Python code using rpy2 and run it through Scalable -============================================================================ - -While using scalable, there may be multiple instances where a model which needs -to be scaled is written in R or can only be interacted with in R. While it is -possible to scale R code by writing it all in an Rscript and calling the script -from Python, it is generally recommended to the R code to Python code using -rpy2. This is to allow for better integration of such code with Python and make -it much easier to modify inputs, get precise outputs, and debug the code. - -The full documentation for rpy2 can be found on this -`website `_. In this guide, only the important -methods for converting R code to Python code will be discussed which is all -that is needed in most of the cases. - -Let's look at an example with R code to convert. The -`HELPS `_ package will be used to demonstrate -the conversion. The R code is as follows: - - -.. code-block:: r - - library(HELPS) - - # test - esi.mon <- cal_heat_stress(TempRes = "month", SECTOR = "SUNF_R", HS = WBGT_ESI, YEAR_INPUT = 2024, - "path/hurs_mon_basd_CanESM5_W5E5v2_GCAM_ref_2015.nc", - "path/tas_mon_basd_CanESM5_W5E5v2_GCAM_ref_2015-2100.nc", - "path/rsds_mon_basd_CanESM5_W5E5v2_GCAM_ref_2015.nc") - - - pwc.mon.hothaps <- cal_pwc(WBGT = esi.mon, LHR = LHR_Hothaps, workload = "high") - pwc.hothaps.ann <- monthly_to_annual(input_rack = pwc.mon.hothaps, SECTOR = "SUNF_R") - ctry_pwc <- grid_to_region(grid_annual_value = pwc.hothaps.ann, SECTOR = "SUNF_R", rast_boundary = country_raster) - glu_pwc <- grid_to_region(grid_annual_value = pwc.hothaps.ann, SECTOR = "SUNF_R", rast_boundary = reg_WB_raster) - - -A simple conversion of the above code to Python code using rpy2 is as follows: - -.. code-block:: python - - import rpy2.robjects.packages as rpackages - import rpy2.robjects as robjects - - helps = rpackages.importr('HELPS') - - esi_mon = helps.cal_heat_stress(TempRes = "month", SECTOR = "SUNF_R", HS = helps.WBGT_ESI, YEAR_INPUT = 2024, - a="hurs_mon_basd_CanESM5_W5E5v2_GCAM_ref_2015-2100.nc", - b="tas_mon_basd_CanESM5_W5E5v2_GCAM_ref_2015-2100.nc", - c="rsds_mon_basd_CanESM5_W5E5v2_GCAM_ref_2015-2100.nc") - - pwc_mon_hothaps = helps.cal_pwc(WBGT = esi_mon, LHR = helps.LHR_Hothaps, workload = "high") - pwc_hothaps_ann = helps.monthly_to_annual(input_rack = pwc_mon_hothaps, SECTOR = "SUNF_R") - - country_raster = robjects.r['country_raster'] - ctry_pwc = helps.grid_to_region(grid_annual_value = pwc_hothaps_ann, SECTOR = "SUNF_R", rast_boundary = country_raster) - - reg_WB_raster = robjects.r['reg_WB_raster'] - glu_pwc = helps.grid_to_region(grid_annual_value = pwc_hothaps_ann, SECTOR = "SUNF_R", rast_boundary = reg_WB_raster) - - -The above code is a simple conversion of the R code to Python code using rpy2. -A couple of important things to note are: - -* The HELPS package is imported at ``helps = rpackages.importr('HELPS')``. - This is similar to importing a package in R using ``library(HELPS)``. - Unlike R, all the functions in a package are not automatically imported. - Instead an object is created whose attributes corresponds to the methods in - that package. - -* If there are functions or values in the R code which are either not defined - in any particular package, are defined in the global environment, or it's - simply unknown where exactly they may reside then the ``robjects.r`` object - can be used to access them. For example, in the above code, the - ``country_raster`` and ``reg_WB_raster`` functions are not defined in the - HELPS package, and without any knowledge where they may reside, the - ``robjects.r`` object can be used to access by just looking up the name of - the function or value in the R code. Another example is if ``pi`` is used - in the R code, then it can be accessed in Python using - ``pi = robjects.r['pi']``. - -* A more subtle point to note is that the ``HeatStress`` function in R takes - three positional arguments at the end. These are matched to the ``...`` - argument in the R code. However, no positional arguments should be passed - to the rpy2 function in python to avoid confusion. To make sure that the - extra arguments get matched with ``...`` internally, arbitrary names can - be used for the arguments. In the above code, such arguments are named - ``a``, ``b``, and ``c``. - -These are the only two important things to note when converting R code to -Python code using rpy2 in most cases. The rest of the code can be converted -directly by just replacing the R syntax with Python syntax. - -Running R code through Scalable -------------------------------- - -Running R code on scalable, while ensuring that no unexpected behavior occurs, -can be a little tricky. While making functions having code similar to the above -example can work to run them through scalable, a crucial step should be taken -to avoid any unexpected behavior. This step involves a preload script. Let's -look at another example. Let's say the python functions below are desired to be -run through scalable: - -.. code-block:: python - - def run_annualized_physical_work_capacity(physical_work_capacity_raster): - import rpy2.robjects.packages as rpackages - - helps = rpackages.importr('HELPS') - - # aggregate physical work capacity to annual values and reformat to a data frame - annualized_physical_work_capacity_df = helps.monthly_to_annual( - input_rack = physical_work_capacity_raster, - SECTOR = "SUNF_R" - ) - - return annualized_physical_work_capacity_df - - - def run_country_physical_work_capacity(annualized_physical_work_capacity_df): - import rpy2.robjects.packages as rpackages - import rpy2.robjects as robjects - - helps = rpackages.importr('HELPS') - country_raster = robjects.r['country_raster'] - - # map annual physical work capacity to gridded countries - country_physical_work_capacity_df = helps.grid_to_region( - grid_annual_value = annualized_physical_work_capacity_df, - SECTOR = "SUNF_R", - rast_boundary = country_raster - ) - - return country_physical_work_capacity_df - - -The above functions would run fine in most cases through scalable. However, -since rpy2 launches an implicit R session when it is imported, it is important -to ensure that the R session is launched before the functions are run to -prevent deadlocks or multiple R sessions from being launched. This can be done -with the use of a preload python script. All this script needs to do is import -rpy2 and everything that may be needed in any of the functions. It should look -something like this if the above functions are to be run: - -.. code-block:: python - - def dask_setup(worker): - import rpy2.robjects.packages as rpackages - import rpy2.robjects as r - from rpy2.robjects import StrVector - helps = rpackages.importr('HELPS') - terra = rpackages.importr('terra') - arrow = rpackages.importr('arrow') - worker.imports = {} - worker.imports['HELPS'] = helps - worker.imports['terra'] = terra - worker.imports['arrow'] = arrow - worker.imports['country_raster'] = r.r['country_raster'] - worker.imports['reg_WB_raster'] = r.r['reg_WB_raster'] - worker.imports['StrVector'] = StrVector - -The script above calls the ``dask_setup`` function with the worker argument. -In this function, everything that may be needed from the R environment is -imported and stored in the worker.imports dictionary. This dictionary is then -used to import the necessary objects in the functions. The path to the preload -script just needs to be passed into the ``PreloadScript`` argument in the -scalable ``add_container`` function. The original functions also need to be -modified a little to use the objects created in the preload script. - -The modified functions would look like this: - -.. code-block:: python - - from scalable import * - - def run_annualized_physical_work_capacity(physical_work_capacity_raster): - worker = get_worker() - helps = worker.imports['HELPS'] - - # aggregate physical work capacity to annual values and reformat to a data frame - annualized_physical_work_capacity_df = helps.monthly_to_annual( - input_rack = physical_work_capacity_raster, - SECTOR = "SUNF_R" - ) - - return annualized_physical_work_capacity_df - - def run_country_physical_work_capacity(annualized_physical_work_capacity_df): - worker = get_worker() - helps = worker.imports['HELPS'] - country_raster = worker.imports['country_raster'] - - # map annual physical work capacity to gridded countries - country_physical_work_capacity_df = helps.grid_to_region( - grid_annual_value = annualized_physical_work_capacity_df, - SECTOR = "SUNF_R", - rast_boundary = country_raster - ) - - return country_physical_work_capacity_df - -The above functions can now be run through scalable without any issues. As a -reminder, the ``add_container`` function should be called with the preload -script path passed into the ``preload_script`` argument. It should look -something like this: - -.. code-block:: python - - from scalable import * - - cluster = SlurmCluster( - queue='short', - walltime='02:00:00', - account='GCIMS', - silence_logs=False - ) - - cluster.add_container( - tag='HELPS', - # 1 cpu since R is single-threaded - cpus=1, - memory='8GB', - preload_script='path/to/preload_script.py' - ) - -Ensuring that the above steps are used to run R code through scalable will -prevent any unexpected behavior. The above steps are general and should work -for most cases where R code needs to be run through scalable. - -If any issues arise or any help is needed converting a specific piece of code, -please feel free to open an issue on the scalable github repo -`here `_. \ No newline at end of file