Persistent/Distributed Generation with Crop Example

This example shows how to use the Crop object for disk-based combo running - either for persistent progress or distributed processing.

First let’s define a very simple function, describe it with a Runner and Harvester and set the combos for this first set of runs.

%config InlineBackend.figure_formats = ['svg']
import xyzpy as xyz


def foo(a, b):
    return a + b, a - b


r = xyz.Runner(foo, ['sum', 'diff'])
h = xyz.Harvester(r, data_name='foo_data.h5')

combos = {
    'a': range(0, 10),
    'b': range(0, 10),
}

We could use the harvester to generate data locally. But if we want results to be written to disk, either for persistence or to run them elsewhere, we need to create a Crop.

c = h.Crop(name='first_run', batchsize=5)
c

<Crop(name='first_run', batchsize=5, num_batches=20)>

Sow the combos

A single crop is used for each set of runs/combos, with batchsize setting how many runs should be lumped together (default: 1). We first sow the combos to disk using the Crop:

c.sow_combos(combos)

100%|##########| 100/100 [00:00<00:00, 124128.56it/s]

There is now a hidden directory containing everything the crop needs:

!ls -a

 .				   dask-worker-space
 ..				  'farming example.ipynb'
'basic output example.ipynb'	   .ipynb_checkpoints
'complex output example.ipynb'	  'visualize linear algebra.ipynb'
'crop example.ipynb'		   .xyz-first_run
'dask distributed example.ipynb'

And inside that are folders for the batches and results, the pickled function, and some other dumped settings:

!ls .xyz-first_run/

batches  results  xyz-function.clpkl  xyz-settings.jbdmp

Once sown, we can check the progress of the Crop:

print(c)

/media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples/.xyz-first_run
-------------------------------------------------------------------=========
5 / 20 batches of size 5 completed
[#####               ] : 25.0%

There are a hundred combinations, with a batchsize of 5, yielding 20 batches to be processed.

Hint

As well as combos you can supply cases and constants to sow_combos().

Grow the results

Any python process with access to the sown batches in .xyz-first_run (and the function requirements) can grow the results (you could even zip the folder up and send elsewhere). The process can be run in several ways:

  1. In the .xyz-first_run folder itself, using e.g:

python -c "import xyzpy; xyzpy.grow(i)"  # with i = 1 ... 20

  1. In the current (‘parent’) folder, one then has to used a named crop to differentiate: e.g:

python -c "import xyzpy; crop=xyzpy.Crop(name='fist_run'); xyzpy.grow(i, crop=crop)"

  1. Somewhere else. Then the parent must be specified too, e.g.:

python -c "import xyzpy; crop=xyzpy.Crop(name='first_run', parent_dir='.../xyzpy/docs/examples'); xyzpy.grow(i, crop=crop)"

To fake this happening we can run grow ourselves (this cell could standalone):

import xyzpy
crop = xyzpy.Crop(name='first_run')
for i in range(1, 11):
    xyzpy.grow(i, crop=crop)

xyzpy: loaded batch 1 of first_run.

{'a': 0, 'b': 4}: 100%|##########| 5/5 [00:00<00:00, 3460.65it/s]

xyzpy: success - batch 1 completed.
xyzpy: loaded batch 2 of first_run.

{'a': 0, 'b': 9}: 100%|##########| 5/5 [00:00<00:00, 4195.14it/s]

xyzpy: success - batch 2 completed.
xyzpy: loaded batch 3 of first_run.

{'a': 1, 'b': 4}: 100%|##########| 5/5 [00:00<00:00, 2392.64it/s]

xyzpy: success - batch 3 completed.
xyzpy: loaded batch 4 of first_run.

{'a': 1, 'b': 9}: 100%|##########| 5/5 [00:00<00:00, 4371.80it/s]

xyzpy: success - batch 4 completed.
xyzpy: loaded batch 5 of first_run.

{'a': 2, 'b': 4}: 100%|##########| 5/5 [00:00<00:00, 4623.35it/s]

xyzpy: success - batch 5 completed.
xyzpy: loaded batch 6 of first_run.

{'a': 2, 'b': 9}: 100%|##########| 5/5 [00:00<00:00, 3317.75it/s]

xyzpy: success - batch 6 completed.
xyzpy: loaded batch 7 of first_run.

{'a': 3, 'b': 4}: 100%|##########| 5/5 [00:00<00:00, 3296.37it/s]

xyzpy: success - batch 7 completed.
xyzpy: loaded batch 8 of first_run.

{'a': 3, 'b': 9}: 100%|##########| 5/5 [00:00<00:00, 4092.00it/s]

xyzpy: success - batch 8 completed.
xyzpy: loaded batch 9 of first_run.

{'a': 4, 'b': 4}: 100%|##########| 5/5 [00:00<00:00, 2017.66it/s]

xyzpy: success - batch 9 completed.
xyzpy: loaded batch 10 of first_run.

{'a': 4, 'b': 9}: 100%|##########| 5/5 [00:00<00:00, 4612.17it/s]

xyzpy: success - batch 10 completed.

And now we can check the progress:

print(c)

/media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples/.xyz-first_run
-------------------------------------------------------------------=========
10 / 20 batches of size 5 completed
[##########          ] : 50.0%

If we were on a batch system we could use xyzpy.Crop.grow_cluster() to automatically submit all missing batches as jobs. It is worth double checking the script that is used first though! This is done using xyzpy.Crop.gen_cluster_script():

print(c.gen_cluster_script(scheduler='sge', minutes=20, gigabytes=1))

#!/bin/bash -l
#$ -S /bin/bash
#$ -N first_run
#$ -l h_rt=0:20:0,mem=1G
#$ -l tmpfs=1G
mkdir -p /home/johnnie/Scratch/output
#$ -wd /home/johnnie/Scratch/output
#$ -pe smp None

#$ -t 1-10
echo 'XYZPY script starting...'
cd /media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples
export OMP_NUM_THREADS=None
export MKL_NUM_THREADS=None
export OPENBLAS_NUM_THREADS=None
export NUMBA_NUM_THREADS=None

conda activate py311
read -r -d '' SCRIPT << EOM
#
from xyzpy.gen.cropping import grow, Crop
if __name__ == '__main__':
    crop = Crop(name='first_run', parent_dir='/media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples')
    print('Growing:', repr(crop))
    grow_kwargs = dict(crop=crop, debugging=False, num_workers=None)
    batch_ids = (11, 12, 13, 14, 15, 16, 17, 18, 19, 20)]
    grow(batch_ids[$SGE_TASK_ID - 1], **grow_kwargs)
EOM
python -c "$SCRIPT"
echo 'XYZPY script finished'

The default scheduler is 'sge' (Sun Grid Engine), however you can also specify 'pbs' (Portable Batch System) or 'slurm':

print(c.gen_cluster_script(scheduler='pbs', minutes=20, gigabytes=1))

#!/bin/bash -l
#PBS -N first_run
#PBS -lselect=None:ncpus=None:mem=1gb
#PBS -lwalltime=00:20:00

#PBS -J 1-10
echo 'XYZPY script starting...'
cd /media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples
export OMP_NUM_THREADS=None
export MKL_NUM_THREADS=None
export OPENBLAS_NUM_THREADS=None
export NUMBA_NUM_THREADS=None

conda activate py311
read -r -d '' SCRIPT << EOM
#
from xyzpy.gen.cropping import grow, Crop
if __name__ == '__main__':
    crop = Crop(name='first_run', parent_dir='/media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples')
    print('Growing:', repr(crop))
    grow_kwargs = dict(crop=crop, debugging=False, num_workers=None)
    batch_ids = (11, 12, 13, 14, 15, 16, 17, 18, 19, 20)
    grow(batch_ids[$PBS_ARRAY_INDEX - 1], **grow_kwargs)
EOM
python -c "$SCRIPT"
echo 'XYZPY script finished'

print(c.gen_cluster_script(scheduler='slurm', minutes=20, gigabytes=1))

#!/bin/bash -l
#SBATCH --job-name=first_run
#SBATCH --time=00:20:00
#SBATCH --mem=1G
#SBATCH --array=1-10
echo 'XYZPY script starting...'
cd /media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples
export OMP_NUM_THREADS=None
export MKL_NUM_THREADS=None
export OPENBLAS_NUM_THREADS=None
export NUMBA_NUM_THREADS=None

conda activate py311
read -r -d '' SCRIPT << EOM
#
from xyzpy.gen.cropping import grow, Crop
if __name__ == '__main__':
    crop = Crop(name='first_run', parent_dir='/media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples')
    print('Growing:', repr(crop))
    grow_kwargs = dict(crop=crop, debugging=False, num_workers=None)
    batch_ids = (11, 12, 13, 14, 15, 16, 17, 18, 19, 20)
    grow(batch_ids[$SLURM_ARRAY_TASK_ID - 1], **grow_kwargs)
EOM
python -c "$SCRIPT"
echo 'XYZPY script finished'

If you are just using the Crop as a persistence mechanism, then xyzpy.Crop.grow() or xyzpy.Crop.grow_missing() will process the batches in the current process:

c.grow_missing(parallel=True)  #  this accepts combo_runner kwargs

100%|##########| 10/10 [00:00<00:00, 19.46it/s]

print(c)

/media/johnnie/Storage2TB/Sync/dev/python/xyzpy/docs/examples/.xyz-first_run
-------------------------------------------------------------------=========
20 / 20 batches of size 5 completed
[####################] : 100.0%

Hint

If different function calls might take different amounts of time based on their arguments, you can supply shuffle=True to xyzpy.Crop.sow_combos(). Each batch will then be a random selection of cases, which should even out the effort each takes as long as batchsize is not too small.

Reap the results

The final step is to ‘reap’ the results from disk. Because the crop was instantiated from a Harvester, that harvester will be automatically used to collect the resulting dataset and sync it with the on-disk dataset:

c.reap()

100%|##########| 100/100 [00:00<00:00, 193196.87it/s]

<xarray.Dataset>
Dimensions:  (a: 10, b: 10)
Coordinates:
  * a        (a) int64 0 1 2 3 4 5 6 7 8 9
  * b        (b) int64 0 1 2 3 4 5 6 7 8 9
Data variables:
    sum      (a, b) int64 0 1 2 3 4 5 6 7 8 9 1 ... 9 10 11 12 13 14 15 16 17 18
    diff     (a, b) int64 0 -1 -2 -3 -4 -5 -6 -7 -8 -9 1 ... 9 8 7 6 5 4 3 2 1 0

Hint

If the Crop is incomplete but has some results, you can call crop.reap(allow_incomplete=True) to harvest the existing data.

Hint

You can supply other kwargs related to harvesting such as overwrite=True, which is useful when you want to replace existing data with newer runs without starting over.

The dataset foo_data.h5 should be on disk, and the crop folder cleaned up:

!ls -a

 .				   dask-worker-space
 ..				  'farming example.ipynb'
'basic output example.ipynb'	   foo_data.h5
'complex output example.ipynb'	   .ipynb_checkpoints
'crop example.ipynb'		  'visualize linear algebra.ipynb'
'dask distributed example.ipynb'

And we can inspect the results:

h.full_ds.xyz.plot(x='a', y='b', z='diff');
../_images/73afdc7df2c31b70dadd8be77413556494d66024dd802a0e326ca17eda16cb3a.svg

Many crops can be created from the harvester at once, and when they are reaped, the results should be seamlessly combined into the on-disk dataset.

# for now clean up
h.delete_ds()