Repository: pangeo-forge/pangeo-forge-recipes (language: python)

Issue:
Dask distributed client memory overruns on Prefect with large time dimensions.
I performed some successful initial tests processing the [noaa-oisst](https://github.com/pangeo-forge/staged-recipes/pull/36) recipe using a smaller time dimension. When extending the recipe to use the full time range `pd.date_range("1981-09-01", "2021-01-05", freq="D")` I am seeing memory overruns on the container where the Prefect Flow is being executed (the container is killed for exceeding its 30GB hard limit).  

To obtain some more details I attempted to run the Flow in a local container directly using `flow.run` rather than the Prefect Agent's subprocess.  I added some `tracemalloc` instrumentation and found rapidly growing memory utilization from `msgpack`.
```
INFO:root:/srv/conda/envs/notebook/lib/python3.8/site-packages/msgpack/__init__.py:0: size=66.9 MiB, count=6, average=11.2 MiB
INFO:root:/srv/conda/envs/notebook/lib/python3.8/site-packages/msgpack/__init__.py:0: size=1156 MiB, count=120, average=9868 KiB
INFO:root:/srv/conda/envs/notebook/lib/python3.8/site-packages/msgpack/__init__.py:0: size=2408 MiB, count=251, average=9826 KiB
INFO:root:/srv/conda/envs/notebook/lib/python3.8/site-packages/msgpack/__init__.py:0: size=3603 MiB, count=377, average=9787 KiB
```
Before the process was killed after exceeding the container's 5GB hard limit.

A traceback of the largest block prior to the container being killed points to message serialization in distributed
```
INFO:root:34 memory blocks: 1448426.9 KiB
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/threading.py", line 890
INFO:root:    self._bootstrap_inner()
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/threading.py", line 932
INFO:root:    self.run()
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/threading.py", line 870
INFO:root:    self._target(*self._args, **self._kwargs)
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/concurrent/futures/thread.py", line 80
INFO:root:    work_item.run()
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/concurrent/futures/thread.py", line 57
INFO:root:    result = self.fn(*self.args, **self.kwargs)
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/utils.py", line 1471
INFO:root:    return await loop.run_in_executor(_offload_executor, lambda: fn(*args, **kwargs))
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/comm/utils.py", line 31
INFO:root:    protocol.dumps(
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/protocol/core.py", line 70
INFO:root:    frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
INFO:root:  File "/srv/conda/envs/notebook/lib/python3.8/site-packages/msgpack/__init__.py", line 35
INFO:root:    return Packer(**kwargs).pack(o)
Killed
```
As a note, my scheduler and workers continue operating fine and processing tasks within their smaller memory limits until the Prefect flow runner is killed.

I've engaged with Prefect support on Slack but this appears to be a potential distributed client issue and I don't know about the depth of their dask expertise now that Jim will be leaving.  I'm very inexperienced with dask internals so any suggestions or advice here would be greatly appreciated 🙇  cc @rabernat @TomAugspurger .


Discussion / hints:
Do you know roughly where in the pipeline things are failing? I assume in `store_chunk`?
@TomAugspurger It is difficult to isolate as it appears that there is no failure of the recipe code in the workers but that the memory increase and failure is occurring in the container where the Prefect flow execution is running (where tasks are being submitted to the cluster by the Prefect [DaskExecutor](https://github.com/PrefectHQ/prefect/blob/master/src/prefect/executors/dask.py#L372)).  

The workers continue processing recipe tasks until the flow execution container is killed.  In the example above where I am using a small 5GB limit for the flow execution container, the workers are only able to [check a single cached input](https://github.com/pangeo-forge/pangeo-forge-recipes/blob/0.3.3/pangeo_forge_recipes/recipes/xarray_zarr.py#L320) before the flow execution container is killed and stops submitting jobs to the cluster.
Ahh, I missed that it's the running the flow that's being killed (rather than a worker). Sorry.

There's been some issues in the past around logs filling out memory on the *scheduler*. Do you know what level logging is set at, and whether anything is being stored in an in-memory list?


> I am seeing memory overruns on the container where the Prefect Flow is being executed (the container is killed for exceeding its 30GB hard limit).

This sounds __exactly__ like #116.

Edit: not exactly. In #116, we are are using the Dask executor, and here we are using prefect.
@rabernat Thanks for seeing the connection here, I hadn't reviewed #116 as I was considering this as a purely Prefect issue.  My (albeit incomplete) understanding of Prefect's DaskExecutor is that no task graph is constructed for the entire flow but that the individual tasks are submitted to the dask scheduler and graph dependencies are managed purely by Prefect.  So in the Prefect case, improving `FilePattern` serialization may fix our issue.  My next steps for testing are to run a test using https://github.com/pangeo-forge/pangeo-forge-recipes/pull/117 and also migrate our Prefect Flow storage to use the script based model https://github.com/pangeo-forge/pangeo-forge-prefect/issues/9 which might also alleviate serialization issues.
I updated the Prefect environment to use `0.3.4`.  Rather than seeing OOM errors on the container where the dask `client` is running and submitting tasks to the `scheduler`, I'm now seeing OOM errors directly on the `scheduler`.
Thanks for your persistence @sharkinsspatial! Such a frustrating situation. Do the logs give you any information about where in the code we are at when we run out of memory?
I'm including some more detailed memory instrumentation for the `scheduler` and `workers` that will hopefully shed some light and trying to dig a bit through the Prefect codebase to understand the type of dask graph which might be constructed for Flow execution.  I'll hopefully have some more details tomorrow but my gut tells me this is most likely closely related to #116.
@sharkinsspatial let me know if you want to do a video call tomorrow to walk through it. I have some free time before 10:00 central and maybe from 3:00 - 4:00 Central.
@TomAugspurger Based on our discussion yesterday I ran simplified flow to try and isolate potential points for memory leaks at the worker level and to try to determine if recipe serialization or something within the recipe class may be responsible for increasing memory usage.

```
def source_url(day: str) -> str:
    day = pd.Timestamp(day)
    source_url_pattern = (
        "https://www.ncei.noaa.gov/data/"
        "sea-surface-temperature-optimum-interpolation/v2.1/access/avhrr/"
        "{day:%Y%m}/oisst-avhrr-v02r01.{day:%Y%m%d}.nc"
    )
    return source_url_pattern.format(day=day)


@task
def download(source_url, cache_location):
    #  tracemalloc.start(25)
    #  snapshot = tracemalloc.take_snapshot()
    target_url = os.path.join(cache_location, str(hash(source_url)))

    try:
        fsspec.open(target_url).open()
        return target_url
    except FileNotFoundError:
        pass

    with fsspec.open(source_url, mode="rb") as source:
        with fsspec.open(target_url, mode="wb") as target:
            target.write(source.read())

    #  top_stats = snapshot.statistics('traceback')
    return target_url


with Flow(
    name,
    storage=storage.S3(bucket="pangeo-forge-aws-bakery-flowstoragebucketpangeof-71w6gsnambj9"),
    run_config=run_config,
    executor=dask_executor,
) as flow:
    days = Parameter(
        "days",
        default=pd.date_range("1981-09-01", "2021-01-05", freq="D").strftime("%Y-%m-%d").tolist(),
    )
    sources = list(map(source_url, pd.date_range("1981-09-01", "2021-01-05", freq="D").strftime("%Y-%m-%d").to_list()))
    nc_sources = download.map(
        sources,
        cache_location=unmapped(f"s3://pangeo-forge-aws-bakery-flowcachebucketpangeofor196cpck7y0pbl/{name}/cache"),
    )
```
Though the rate of memory increase at the worker level is lower, there is still a steady, linear memory increase on workers.  Looking at the tops stats from `tracemalloc` `distributed` is increasing memory allocation with each task that is run
```
/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/protocol/core.py:105: size=2149 MiB (+557 MiB), count=7616 (+1790), average=289 KiB
```
```
/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/protocol/core.py:105: size=2659 MiB (+443 MiB), count=9261 (+1425), average=294 KiB
```

Looking at the `traceback` for the highest `stat` reports
```
8882 memory blocks: 2579912.4 KiB
--
File "<string>", line 1
File "/srv/conda/envs/notebook/lib/python3.8/multiprocessing/spawn.py", line 116
exitcode = _main(fd, parent_sentinel)
File "/srv/conda/envs/notebook/lib/python3.8/multiprocessing/spawn.py", line 129
return self._bootstrap(parent_sentinel)
File "/srv/conda/envs/notebook/lib/python3.8/multiprocessing/process.py", line 315
self.run()
File "/srv/conda/envs/notebook/lib/python3.8/multiprocessing/process.py", line 108
self._target(*self._args, **self._kwargs)
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/process.py", line 192
target(*args, **kwargs)
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/nanny.py", line 792
loop.run_sync(run)
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/tornado/ioloop.py", line 524
self.start()
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/tornado/platform/asyncio.py", line 199
self.asyncio_loop.run_forever()
File "/srv/conda/envs/notebook/lib/python3.8/asyncio/base_events.py", line 570
self._run_once()
File "/srv/conda/envs/notebook/lib/python3.8/asyncio/base_events.py", line 1859
handle._run()
File "/srv/conda/envs/notebook/lib/python3.8/asyncio/events.py", line 81
self._context.run(self._callback, *self._args)
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/worker.py", line 999
await self.handle_stream(
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/core.py", line 555
msgs = await comm.read()
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/comm/tcp.py", line 218
msg = await from_frames(
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/comm/utils.py", line 79
res = _from_frames()
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/comm/utils.py", line 62
return protocol.loads(
File "/srv/conda/envs/notebook/lib/python3.8/site-packages/distributed/protocol/core.py", line 105
return msgpack.loads(
```

In reference to https://github.com/dask/distributed/issues/4091 I have decreased profiler frequency using these environment settings on the `scheduler` and `worker` containers 
```
"DASK_DISTRIBUTED__WORKER__PROFILE__INTERVAL": "10000ms",
"DASK_DISTRIBUTED__WORKER__PROFILE__CYCLE": "1000000ms",
```
But this does not seem to alleviate the issue.

Interesting, thanks... Do you think it's worth removing fsspec and s3fs from the equation now, and just using requests and boto3 directly?
Could the memory growth be related to fsspec / s3fs caching? If so, that can be turned off.
We discussed that briefly yesterday. Looking at the docs, we felt like it should have an upper bound and start purge items from the cache by default, but didn't test out disabling it yet.
@rabernat  I had previously tested using 
```
fs = S3FileSystem(
        anon=False,
        default_cache_type="none",
        default_fill_cache=False,
    )
```

In order to see if s3fs caching could be playing apart that results were the same.  In order to drastically simplify things and potentially isolate this to a Dask or Prefect issue I made another run using 
```
@task
def download(source_url, cache_location):
    time.sleep(0.4)
```
Which showed the same steadily increasing memory usage.  To my naive eye, this coupled with the `tracemalloc` output showing steadily increasing memory use of `distributed` indicates that this is some type of issue with the how Prefect dask `client` is submitting tasks or a lower core issue in `distributed`.  I am quite far out of my depth trying to diagnose any core issue here. I've referenced this issue the Prefect Slack for communications with their developer team.  On the dask side of things, @TomAugspurger what would be the next steps for us to add further details and escalate https://github.com/dask/distributed/issues/4091?  There are also several loosely related issues but this one seems to be the most relevant to this case.
@TomAugspurger A discussion of a similar issue from the [Prefect Slack channel] (https://prefect-community.slack.com/archives/CL09KU1K7/p1623226992151500) references this older issue you raised on [distributed](https://github.com/dask/distributed/issues/2757). 
Picking up the thread on *worker* memory issues,  https://gist.github.com/TomAugspurger/f870d87456695cf47bafd5a8a5ae450e as an incomplete start at a worker plugin to periodically snapshot the types & sizes of all the objects in the worker process's memory. That might shed some light on which objects are sticking around.

I don't know if I'll have a chance to get back to that today or tomorrow.
As noted in #208 there are Prefect specific task serialization issues that are being tracked by Prefect with ongoing work [here](https://github.com/PrefectHQ/prefect/pull/5004).  Given this issue I have been running recipes directly without Prefect.  Testing with a reduced time dimension version of the [gpm-imerg recipe](https://github.com/pangeo-forge/gpm-imerge-hhr-feedstock) I have also encountered potential serialization related issues. I realize I haven't included easily reproducible examples of problematic recipe serialization for large time dimensions.  Using the following recipe code
```
import aiohttp
import pandas as pd

from pangeo_forge_recipes.patterns import ConcatDim, FilePattern
from pangeo_forge_recipes.recipes import XarrayZarrRecipe

# TODO: replace with ENV vars
username = password = "pangeo@developmentseed.org"


def make_filename(time):
    input_url_pattern = (
        "https://arthurhouhttps.pps.eosdis.nasa.gov/gpmdata/{yyyy}/{mm}/{dd}/"
        "imerg/3B-HHR.MS.MRG.3IMERG.{yyyymmdd}-S{sh}{sm}00-E{eh}{em}59.{MMMM}.V06B.HDF5"
    ).format(
        yyyy=time.strftime("%Y"),
        mm=time.strftime("%m"),
        dd=time.strftime("%d"),
        yyyymmdd=time.strftime("%Y%m%d"),
        sh=time.strftime("%H"),
        sm=time.strftime("%M"),
        eh=time.strftime("%H"),
        em=(time + pd.Timedelta("29 min")).strftime("%M"),
        MMMM=f"{(time.hour*60 + time.minute):04}",
    )
    return input_url_pattern


dates = pd.date_range("2020-05-31T00:00:00", "2021-05-31T23:59:59", freq="30min")
time_concat_dim = ConcatDim("time", dates, nitems_per_file=1)
pattern = FilePattern(
    make_filename,
    time_concat_dim,
    fsspec_open_kwargs={"auth": aiohttp.BasicAuth(username, password)},
)

recipe = XarrayZarrRecipe(
    pattern,
    xarray_open_kwargs={"group": "Grid", "decode_coords": "all"},
    inputs_per_chunk=1,
)
```
With a `LocalCluster`
```
from distributed import Client, LocalCluster
from distributed.worker import Worker
from pangeo_forge_recipes.storage import CacheFSSpecTarget, FSSpecTarget, MetadataTarget
import dask
from pympler import asizeof
import tempfile
from fsspec.implementations.local import LocalFileSystem
from imerg_recipe import recipe

fs_local = LocalFileSystem()
cache_dir = tempfile.TemporaryDirectory()
cache_target = CacheFSSpecTarget(fs_local, cache_dir.name)
target_dir = tempfile.TemporaryDirectory()
target = FSSpecTarget(fs_local, target_dir.name)
metadata_dir = tempfile.TemporaryDirectory()
metadata_target = MetadataTarget(fs_local, metadata_dir.name)

recipe.target = target
recipe.input_cache = cache_target
recipe.metadata_cache = metadata_target
delayed = recipe.to_dask()
print(dask.utils.format_bytes(asizeof.asizeof(delayed)))


import tracemalloc
tracemalloc.start(15)


def memory_logger():
    import time
    while True:
        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics("filename")
        if len(top_stats) > 0:
            stat = top_stats[0]
            print("%s memory blocks: %.1f MB" % (stat.count, stat.size / (1024 * 1024)))
            for line in stat.traceback.format():
                print(line)
        time.sleep(5)

import threading
thread = threading.Thread(target=memory_logger)
thread.daemon = True
thread.start()

with LocalCluster(
    n_workers=2,
    worker_class=Worker,
    memory_limit='5GB',
) as cluster, Client(cluster) as client:
    delayed.compute()
    cluster.close()
```
Our 5GB worker limit is quickly overrun prior to any task being executed.  I don't have a strong understanding of task serialization in Dask but I'm curious how this recipe which is reported 21.36MB causes such a large memory footprint on both the client and workers.
https://github.com/pangeo-forge/pangeo-forge-recipes/issues/151#issuecomment-931607965 ☝️ cc @alxmrs
Interestingly, in our CI based workflows we don't have access to the recipe module location at runtime and instead interrogate the [meta.yaml](https://github.com/pangeo-forge/roadmap/blob/master/doc/adr/0002-use-meta-yaml-to-track-feedstock-metadata.md) to dynamically import the the recipe module.
```
import os
import importlib


name = "recipe"
module_path = os.path.abspath("./imerg_recipe.py")
spec = importlib.util.spec_from_file_location(name, module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
recipe = getattr(module, name)
```
This appears to cause Dask to use a different serialization strategy.  In this case we see a slow memory expansion (led by `cloudpickle_fast.py`) on the client thread and data is transferred to the workers at a much slower rate but still resulting in the same eventual worker memory overrun.  I don't have a clear picture of the interplay between module import method and pickling and why this approach results in different behavior.

@sharkinsspatial thanks for posting that. Just to clarify, when you run the job do you see the memory spike on the scheduler, workers, or both?

And do you have a sense for what the peak memory usage should be? Something like the size of the largest chunk times the number of chunks being merged?
@TomAugspurger In the first case (with a normal module `import`) there is a rapid spike in memory to ≈5GB on the client followed by almost immediate memory growth on the workers until they begin reporting 
```
distributed.worker - WARNING - Unmanaged memory use is high. This may indicate a memory leak or the memory may not be released to the OS; see https://distributed.dask.org/en/latest/worker.html#memtrim for more information. -- Unmanaged memory: 4.19 GiB -- Worker memory limit: 4.66 GiB
```
Scheduler memory increases rapidly inline with worker memory growth.  As for peak memory I would expect that your assumption for peak memory is correct, however this memory growth is almost immediate and is occurring prior to any tasks being executed (`cache_inputs` is the first stage and none of the tasks have even registered in the dashboard before the workers are OOM).

In the second case when obtaining the recipe module via 
```
spec.loader.exec_module(module)
```
Memory growth slowly increases on both the client, workers and scheduler until the workers report `Unmanaged memory use is high` as if the serialization and Client->Scheduler data transfer process is much slower when using this module reference strategy.  So, same end result, just occurring much more slowly.
I've narrowed the serialization issue down to just `FilePattern`. I'll post more details later (have to run again), but the gist is if we replace every element of the Dask graph with "simple" objects, then things are fine.

The code below completes without any issues (doesn't do anything other than submit a bunch of no-op tasks to dask)

<details>

```python
import dataclasses
import functools
import tempfile
import dask.sizeof
import dask.utils
from distributed import Client
from fsspec.implementations.local import LocalFileSystem

import pangeo_forge_recipes.patterns
from imerg_recipe import recipe
from pangeo_forge_recipes.storage import CacheFSSpecTarget, FSSpecTarget, MetadataTarget


# @dask.sizeof.sizeof.register(pangeo_forge_recipes.patterns.FilePattern)
# def sizeof_file_pattern(o: pangeo_forge_recipes.patterns.FilePattern) -> int:
#     return (
#         dask.sizeof.sizeof(o.format_func) +
#         dask.sizeof.sizeof(o.fsspec_open_kwargs) +
#         dask.sizeof.sizeof(o.query_string_secrets) +
#         dask.sizeof.sizeof(o.is_opendap)
#     )


# @dask.sizeof.sizeof.register(pangeo_forge_recipes.patterns.MergeDim)
# @dask.sizeof.sizeof.register(pangeo_forge_recipes.patterns.ConcatDim)
# def sizeof_combinedim(o) -> int:
#     return dask.sizeof.sizeof(dataclasses.asdict(o))


fs_local = LocalFileSystem()
cache_dir = tempfile.TemporaryDirectory()
cache_target = CacheFSSpecTarget(fs_local, cache_dir.name)
target_dir = tempfile.TemporaryDirectory()
target = FSSpecTarget(fs_local, target_dir.name)
metadata_dir = tempfile.TemporaryDirectory()
metadata_target = MetadataTarget(fs_local, metadata_dir.name)

recipe.target = target
recipe.input_cache = cache_target
recipe.metadata_cache = metadata_target
obj = recipe.to_dask()
dsk = obj.dask


def simple_func(*args, **kwargs):
    return None


def simplify_callable(dsk):
    for k, v in dsk.items():
        _, *args = v
        dsk[k] = (simple_func,) + tuple(args)


def simplify_args(dsk):
    simple_types = (int, str, list, pangeo_forge_recipes.patterns.Index)
    for k, v in dsk.items():
        func, *args = v
        new_args = []
        n = len(args)
        for arg in args:
            if isinstance(arg, simple_types):
                new_args.append(arg)
            elif isinstance(arg, functools.partial):
                replace = {"file_pattern"}
                # replace = {}
                new_kwargs = dict()
                for i, (kwarg, val) in enumerate(arg.keywords.items()):
                    if kwarg in replace:
                        new_kwargs[kwarg] = i
                    else:
                        new_kwargs[kwarg] = val

                new_arg = functools.partial(arg.func, **new_kwargs)
                new_args.append(new_arg)
            else:
                raise ValueError

        assert len(new_args) == n
        dsk[k] = (func,) + tuple(new_args)


if __name__ == "__main__":
    print("simplifying")
    simplify_callable(dsk)
    simplify_args(dsk)

    with Client(n_workers=1, threads_per_worker=1, memory_limit="4G", nanny=False) as client:
        print("submitting")
        client.get(dsk, [obj.key])

```

</details>


If we *don't* replace the `file_pattern` objects in the values of the partially applied functions in the arguments in the Dask graph (what a sentence), then I see the memory issues (on the worker and scheduler). You can do that by changing `replace = {"file_pattern"}` to `replace = {}` and running the program again.

I'll pick this up again this afternoon.
https://github.com/pangeo-forge/pangeo-forge-recipes/pull/217 might have a fix. @sharkinsspatial are you able to test out the full recipe on a branch?
@TomAugspurger This appears to solve our initial memory overrun (though I can't fully test as the NASA server bounces me with `aiohttp.client_exceptions.ServerDisconnectedError` almost immediately 🤣 ).  Let me try to test on a k8s cluster as well.  My one concern there is that (because of what I assume are network communication limitations between my client and the scheduler in the cluster) I hit OOM errors on the client pod as the graph is buffered during the initial call to `compute`.  Out of curiosity, do we have a sense of why the serialized `FilePattern` are so large?
@sharkinsspatial as I noted in https://github.com/pangeo-forge/pangeo-forge-recipes/pull/217#issuecomment-932432548, my "fix" might be off. I'm not sure anymore.

I do have something that I think *is* a fix on the recipe though. Can you try with this for the `imerg_recipe.py`? No changes to pangeo-forge-recipes.

```python
import aiohttp
import pandas as pd

from pangeo_forge_recipes.patterns import ConcatDim, FilePattern
from pangeo_forge_recipes.recipes import XarrayZarrRecipe

# TODO: replace with ENV vars
username = password = "pangeo@developmentseed.org"


def make_filename(time):
    input_url_pattern = (
        "https://arthurhouhttps.pps.eosdis.nasa.gov/gpmdata/{yyyy}/{mm}/{dd}/"
        "imerg/3B-HHR.MS.MRG.3IMERG.{yyyymmdd}-S{sh}{sm}00-E{eh}{em}59.{MMMM}.V06B.HDF5"
    ).format(
        yyyy=time.strftime("%Y"),
        mm=time.strftime("%m"),
        dd=time.strftime("%d"),
        yyyymmdd=time.strftime("%Y%m%d"),
        sh=time.strftime("%H"),
        sm=time.strftime("%M"),
        eh=time.strftime("%H"),
        em=(time + pd.Timedelta("29 min")).strftime("%M"),
        MMMM=f"{(time.hour*60 + time.minute):04}",
    )
    return input_url_pattern


dates = pd.date_range("2020-05-31T00:00:00", "2021-05-31T23:59:59", freq="30min")

def func(i):
    return pd.date_range("2020-05-31T00:00:00", "2021-05-31T23:59:59", freq="30min")[i]


class SymbolicKeys:
    def __init__(self, func, n):
        self.func = func
        self.n = n

    def __iter__(self):
        for i in range(self.n):
            yield self.func(i)

    def __len__(self):
        return self.n



keys = SymbolicKeys(func, len(dates))


time_concat_dim = ConcatDim("time", keys, nitems_per_file=1)
pattern = FilePattern(
    make_filename,
    time_concat_dim,
    fsspec_open_kwargs={"auth": aiohttp.BasicAuth(username, password)},
)

recipe = XarrayZarrRecipe(
    pattern,
    xarray_open_kwargs={"group": "Grid", "decode_coords": "all"},
    inputs_per_chunk=1,
)
```

The difference is that `SymbolicKeys` bit, which avoids ever putting the large DatetimeIndex of keys in the task graph.

Happy to jump on https://whereby.com/pangeo to chat through this sometime in the next 30 minutes.


> The difference is that SymbolicKeys bit, which avoids ever putting the large DatetimeIndex of keys in the task graph.

Cool idea. If this solution sticks, we could consider having `ConcatDim.__post_init__` make this transformation for users.
I think a more general solution to this might be to just never ship the `FilePattern` to the tasks at all. There is no fundamental reason why all tasks need a copy of the index.
Yeah, I think that's the right solution if it's doable. Looking now.
It's not a trivial refactor but should be possible.
I can try to make that PR.
That'd be great. I'm getting a bit tripped up how to get `fname = file_pattern[input_key]` early enough that it doesn't get into the task graph. I started at https://github.com/TomAugspurger/pangeo-forge/pull/new/fix/151-memory-filepattern (currently broken).

And I'm like 90% sure that this SymbolicKeys stuff will fix the real recipe, but maybe we want to wait to hear from @sharkinsspatial before doing another large refactor.
I do think the SymbolicKeys will fix it. But I think that the SymbolicKeys approach adds complexity, while mine removes it. Let's try both. I'm going to add this to #213, which is already about how we open files.
@TomAugspurger This `SymbolicKeys` approach is failing with 
```
Function:  cache_input
args:      ((DimIndex(name='time', index=5958, sequence_len=17568, operation=<CombineOp.CONCAT: 2>),))
kwargs:    {}
Exception: TypeError("'SymbolicKeys' object is not subscriptable")
```
For each of the time indice tasks.
@sharkinsspatial can you try with

```python
class SymbolicKeys:
    def __init__(self, func, n):
        self.func = func
        self.n = n

    def __iter__(self):
        for i in range(self.n):
            yield self.func(i)

    def __len__(self):
        return self.n

    def __getitem__(self, i):
        return self.func(i)
```

We need to figure out exactly what features of that `ConcatDim.keys` we were using in the actual functions. I'll take a look, but you can try this in the meantime if it's easy.

According to the type signature, it just needs to be a collection (so `__getitem__` and `__len__`) so maybe that implementation will work.
@TomAugspurger I had actually just tried that and hit 
```
distributed.worker - WARNING - Compute Failed
Function:  cache_input
args:      ((DimIndex(name='time', index=8785, sequence_len=17568, operation=<CombineOp.CONCAT: 2>),))
kwargs:    {}
Exception: AttributeError("'Timestamp' object has no attribute 'strftie'")
```
Is the error message actually `strftie`, because if so that's true. It should be `strftime`, so there's a typo somewhere.
@TomAugspurger 👍  Yup. Our `em` pattern above was incorrect `em=(time + pd.Timedelta("29 min")).strftime("%M")`.   This runs cleanly with flat memory utilization on the Client, Scheduler and Workers.  I'll make a more full scale run on the k8s cluster to verify.
So in terms of "root"-cause it seems like passing around `ConcatDim` objects, specifically large `ConcatDim.keys`, is a bad idea. I'm not 100% sure why, but my guess is that somewhere when deserializing tasks, dask ends up creating `n_inputs` copies of the DatetimeIndex. In this recipe that's something like:

```python
In [20]: dask.utils.format_bytes(len(dates) * dates.nbytes)
Out[20]: '2.30 GiB'
```

I think that on the client, we have ~17,000 references to the same `dates` DatetimeIndex, while on the scheduler / worker we have ~17,000 individual DatetimeIndexes. That's my guess anyway.

---

Ideally we're able to extract the relevant (small) pieces of data from these FilePattern / ConcatDim objects before passing them to functions so that they don't end up in the task graph. I don't think we want users worrying about SymbolicKeys type things if we can help it.
> it seems like passing around `ConcatDim` objects, specifically large `ConcatDim.keys`, is a bad ide

This makes sense. I'm working on a PR to address this.
@rabernat I'll track the PR work, ping me if there is any integration testing you would like done.  On a somewhat related note, though @TomAugspurger's initial `sizeof` registration https://github.com/pangeo-forge/pangeo-forge-recipes/pull/217 is not the solution here, I do think this would be valuable to the Prefect team in their refactoring work as this situation of task size underestimation causing the scheduler to oversubscribe a worker appears to be similar to their case (though I believe they also have significant work to do in reducing the deserialized size of individual tasks) refs https://github.com/PrefectHQ/prefect/pull/5004

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 0eaa3b15..9d878d3d 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -28,7 +28,7 @@ repos:
     - id: seed-isort-config
 
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: 'v0.812'
+    rev: 'v0.910'
     hooks:
     -   id: mypy
         exclude: tests
diff --git a/docs/development/release_notes.md b/docs/development/release_notes.md
index 94f2e63e..00bee54d 100644
--- a/docs/development/release_notes.md
+++ b/docs/development/release_notes.md
@@ -2,6 +2,11 @@
 
 ## v0.6.1 - Unreleased
 
+- Major internal refactor of executors. {pull}`219`.
+  Began deprecation cycle for recipe methods (e.g. `recipe.prepare_target()`) in
+  favor of module functions.
+- Addition of `open_input_with_fsspec_reference` option on {class}`pangeo_forge_recipes.recipes.XarrayZarrRecipe`,
+  permitting the bypassing of h5py when opening inputs. {pull}`218`
 
 ## v0.6.0 - 2021-09-02
 
diff --git a/pangeo_forge_recipes/executors.py b/pangeo_forge_recipes/executors.py
deleted file mode 100644
index bca10b1b..00000000
--- a/pangeo_forge_recipes/executors.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from rechunker.executors import DaskPipelineExecutor  # noqa: F401
-from rechunker.executors import PrefectPipelineExecutor  # noqa: F401
-from rechunker.executors import PythonPipelineExecutor  # noqa: F401
diff --git a/pangeo_forge_recipes/executors/__init__.py b/pangeo_forge_recipes/executors/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/pangeo_forge_recipes/executors/base.py b/pangeo_forge_recipes/executors/base.py
new file mode 100644
index 00000000..a8867132
--- /dev/null
+++ b/pangeo_forge_recipes/executors/base.py
@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import enum
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, Generic, Iterable, Optional, TypeVar, Union
+
+from mypy_extensions import NamedArg
+
+Config = Any  # TODO: better typing for config
+SingleArgumentStageFunction = Callable[[Any, NamedArg(type=Any, name="config")], None]  # noqa: F821
+NoArgumentStageFunction = Callable[[NamedArg(type=Any, name="config")], None]  # noqa: F821
+StageFunction = Union[NoArgumentStageFunction, SingleArgumentStageFunction]
+
+
+class StageAnnotationType(enum.Enum):
+    CONCURRENCY = enum.auto()
+    RETRIES = enum.auto()
+
+
+StageAnnotations = Dict[StageAnnotationType, Any]
+
+
+@dataclass(frozen=True)
+class Stage:
+    function: StageFunction
+    name: str
+    mappable: Optional[Iterable] = None
+    annotations: Optional[StageAnnotations] = None
+
+
+@dataclass(frozen=True)
+class Pipeline:
+    stages: Iterable[Stage]
+    config: Optional[Config] = None
+
+
+T = TypeVar("T")
+
+
+class PipelineExecutor(Generic[T]):
+    @staticmethod
+    def compile(pipeline: Pipeline) -> T:
+        raise NotImplementedError
+
+    @staticmethod
+    def execute(plan: T):
+        raise NotImplementedError
diff --git a/pangeo_forge_recipes/executors/dask.py b/pangeo_forge_recipes/executors/dask.py
new file mode 100644
index 00000000..b1ec6e9f
--- /dev/null
+++ b/pangeo_forge_recipes/executors/dask.py
@@ -0,0 +1,80 @@
+from typing import Any, Dict, Set, Tuple, Union
+
+import dask
+from dask.delayed import Delayed
+from dask.highlevelgraph import HighLevelGraph
+
+from .base import Pipeline, PipelineExecutor
+
+
+def wrap_map_task(function):
+    # dependencies are dummy args used to create dependence between stages
+    def wrapped(map_arg, config, *dependencies):
+        return function(map_arg, config=config)
+
+    return wrapped
+
+
+def wrap_standalone_task(function):
+    def wrapped(config, *dependencies):
+        return function(config=config)
+
+    return wrapped
+
+
+def checkpoint(*args):
+    return
+
+
+def append_token(task_name: str, token: str) -> str:
+    return f"{task_name}-{token}"
+
+
+class DaskPipelineExecutor(PipelineExecutor[Delayed]):
+    @staticmethod
+    def compile(pipeline: Pipeline):
+
+        token = dask.base.tokenize(pipeline)
+
+        # we are constructing a HighLevelGraph from scratch
+        # https://docs.dask.org/en/latest/high-level-graphs.html
+        layers = dict()  # type: Dict[str, Dict[Union[str, Tuple[str, int]], Any]]
+        dependencies = dict()  # type: Dict[str, Set[str]]
+
+        # start with just the config as a standalone layer
+        # create a custom delayed object for the config
+        config_task = "config"
+        config_token = append_token(config_task, token)
+        layers[config_task] = {config_token: pipeline.config}
+        dependencies[config_task] = set()
+
+        prev_token = ()  # type: Tuple[str, ...]
+        prev_task = config_task
+        for stage in pipeline.stages:
+            stage_token = append_token(stage.name, token)
+            stage_graph = {}  # type: Dict[Union[str, Tuple[str, int]], Any]
+            if stage.mappable is None:
+                func = wrap_standalone_task(stage.function)
+                stage_graph[stage_token] = (func, config_token) + prev_token
+                prev_token = (stage_token,)
+            else:
+                func = wrap_map_task(stage.function)
+                checkpoint_args = []
+                for i, m in enumerate(stage.mappable):
+                    key = (stage.name, i)
+                    stage_graph[key] = (func, m, config_token) + prev_token
+                    checkpoint_args.append(key)
+                checkpoint_key = f"{stage.name}-checkpoint-{token}"
+                stage_graph[checkpoint_key] = (checkpoint, *checkpoint_args)
+                prev_token = (checkpoint_key,)
+            layers[stage.name] = stage_graph
+            dependencies[stage.name] = {prev_task}
+            prev_task = stage.name
+
+        hlg = HighLevelGraph(layers, dependencies)
+        delayed = Delayed(prev_token[0], hlg)
+        return delayed
+
+    @staticmethod
+    def execute(delayed: Delayed):
+        delayed.compute()
diff --git a/pangeo_forge_recipes/executors/function.py b/pangeo_forge_recipes/executors/function.py
new file mode 100644
index 00000000..0d324d6f
--- /dev/null
+++ b/pangeo_forge_recipes/executors/function.py
@@ -0,0 +1,21 @@
+from typing import Callable
+
+from .base import Pipeline, PipelineExecutor
+
+
+class FunctionPipelineExecutor(PipelineExecutor[Callable]):
+    @staticmethod
+    def compile(pipeline: Pipeline):
+        def function():
+            for stage in pipeline.stages:
+                if stage.mappable is not None:
+                    for m in stage.mappable:
+                        stage.function(m, config=pipeline.config)
+                else:
+                    stage.function(config=pipeline.config)
+
+        return function
+
+    @staticmethod
+    def execute(func: Callable) -> None:
+        func()
diff --git a/pangeo_forge_recipes/executors/prefect.py b/pangeo_forge_recipes/executors/prefect.py
new file mode 100644
index 00000000..f7cd887f
--- /dev/null
+++ b/pangeo_forge_recipes/executors/prefect.py
@@ -0,0 +1,45 @@
+from typing import Any, Dict, List, Optional
+
+from prefect import Flow, task, unmapped
+
+from .base import Pipeline, PipelineExecutor, StageAnnotations, StageAnnotationType
+
+
+def annotations_to_task_kwargs(annotations: Optional[StageAnnotations]) -> Dict[str, Any]:
+    if annotations is None:
+        return {}
+    task_kwargs = {}
+    for ann_type in annotations:
+        if ann_type == StageAnnotationType.RETRIES:
+            task_kwargs["max_retries"] = annotations[ann_type]
+        if ann_type == StageAnnotationType.CONCURRENCY:
+            raise ValueError("Haven't figured out concurrency yet.")
+    return task_kwargs
+
+
+class PrefectPipelineExecutor(PipelineExecutor[Flow]):
+    @staticmethod
+    def compile(pipeline: Pipeline):
+
+        with Flow("pangeo-forge-recipe") as flow:
+            upstream_tasks = []  # type: List[task]
+            for stage in pipeline.stages:
+                task_kwargs = annotations_to_task_kwargs(stage.annotations)
+                stage_task = task(stage.function, name=stage.name, **task_kwargs)
+                if stage.mappable is not None:
+                    stage_task_called = stage_task.map(
+                        stage.mappable,
+                        config=unmapped(pipeline.config),
+                        upstream_tasks=[unmapped(t) for t in upstream_tasks],
+                    )
+                else:
+                    stage_task_called = stage_task(
+                        config=pipeline.config, upstream_tasks=upstream_tasks
+                    )
+                upstream_tasks = [stage_task_called]
+
+        return flow
+
+    @staticmethod
+    def execute(flow: Flow):
+        return flow.run()
diff --git a/pangeo_forge_recipes/recipes/base.py b/pangeo_forge_recipes/recipes/base.py
index 3bee9f5c..63316d03 100644
--- a/pangeo_forge_recipes/recipes/base.py
+++ b/pangeo_forge_recipes/recipes/base.py
@@ -1,227 +1,38 @@
-import warnings
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, replace
-from functools import partial
-from typing import Callable, Hashable, Iterable
+from __future__ import annotations
 
-from rechunker.types import MultiStagePipeline, ParallelPipelines, Stage
+from abc import ABC
+from dataclasses import dataclass, replace
+from typing import Callable, ClassVar
 
+from ..executors.base import Pipeline
 from ..patterns import FilePattern, prune_pattern
 
-# How to manually execute a recipe: ###
-#
-#   t = PangeoForgeTarget()
-#   r = MyRecipe(target=t, **opts) # 1
-#   # manual execution of recipe
-#   for input_key in r.iter_inputs():
-#       r.cache_input(input_key) # 4
-#   r.prepare_target() # 3
-#   for chunk_key in r.iter_chunks():
-#       r.store_chunk(chunk_key) # 5
-#   r.finalize_target() # 6
-
-
-# 1) Initialize the Recipe object
-# 2) Point the Recipe at its Target
-# 3) Initialize the recipe.
-#    Check if the target exists; if not, create it.
-# 4) cache the inputs to proximate storage (OPTIONAL)
-#    Some recipes won't need this (e.g. cloud to cloud)
-#    If so, iter_inputs is just an empty iterator
-# 5) Load each chunk from the inputs and store it in the target
-#    Might be coming from the cache or might be read directly.
-# 6)
-
 
+@dataclass
 class BaseRecipe(ABC):
-    """Base recipe class from which all other Recipes inherit.
-    """
-
-    @property
-    @abstractmethod
-    def prepare_target(self) -> Callable[[], None]:
-        """Prepare the recipe for execution by initializing the target.
-        Attribute that returns a callable function.
-        """
-        pass
-
-    @abstractmethod
-    def iter_inputs(self) -> Iterable[Hashable]:
-        """Iterate over all inputs."""
-        pass
-
-    @property
-    @abstractmethod
-    def cache_input(self) -> Callable[[Hashable], None]:
-        """Copy an input from its source location to the cache.
-        Attribute that returns a callable function.
-        """
-        pass
-
-    @abstractmethod
-    def iter_chunks(self) -> Iterable[Hashable]:
-        """Iterate over all target chunks."""
-        pass
-
-    @property
-    @abstractmethod
-    def store_chunk(self) -> Callable[[Hashable], None]:
-        """Store a chunk of data in the target.
-        Attribute that returns a callable function.
-        """
-        pass
+    _compiler: ClassVar[RecipeCompiler]
 
-    @property
-    @abstractmethod
-    def finalize_target(self) -> Callable[[], None]:
-        """Final step to finish the recipe after data has been written.
-        Attribute that returns a callable function.
-        """
-        pass
+    def to_function(self):
+        from ..executors.function import FunctionPipelineExecutor
 
-    def to_pipelines(self) -> ParallelPipelines:
-        """Translate recipe to pipeline for execution.
-        """
-        warnings.warn(
-            "'to_pipelines' is deprecated. Use one of 'to_function', 'to_dask', or "
-            "'to_prefect' directly instead.",
-            FutureWarning,
-        )
-        pipeline = []  # type: MultiStagePipeline
-        # TODO: allow recipes to customize which stages to run
-        pipeline.append(Stage(self.cache_input, list(self.iter_inputs())))
-        pipeline.append(Stage(self.prepare_target))
-        pipeline.append(Stage(self.store_chunk, list(self.iter_chunks())))
-        pipeline.append(Stage(self.finalize_target))
-        pipelines = []  # type: ParallelPipelines
-        pipelines.append(pipeline)
-        return pipelines
-
-    def to_function(self) -> Callable[[], None]:
-        """
-        Translate the recipe to a Python function for execution.
-        """
-
-        def pipeline():
-            # TODO: allow recipes to customize which stages to run
-            for input_key in self.iter_inputs():
-                self.cache_input(input_key)
-            self.prepare_target()
-            for chunk_key in self.iter_chunks():
-                self.store_chunk(chunk_key)
-            self.finalize_target()
-
-        return pipeline
+        return FunctionPipelineExecutor.compile(self._compiler())
 
     def to_dask(self):
-        """
-        Translate the recipe to a dask.Delayed object for parallel execution.
-        """
-        # This manually builds a Dask task graph with each stage of the recipe.
-        # We use a few "checkpoints" to ensure that downstream tasks depend
-        # on upstream tasks being done before starting. We use a manual task
-        # graph rather than dask.delayed to avoid some expensive tokenization
-        # in dask.delayed
-        import dask
-        from dask.delayed import Delayed
-
-        # TODO: HighlevelGraph layers for each of these mapped inputs.
-        # Cache Input --------------------------------------------------------
-        dsk = {}
-        token = dask.base.tokenize(self)
-
-        # TODO: allow recipes to customize which stages to run
-        for i, input_key in enumerate(self.iter_inputs()):
-            dsk[(f"cache_input-{token}", i)] = (self.cache_input, input_key)
-
-        # Prepare Target ------------------------------------------------------
-        dsk[f"checkpoint_0-{token}"] = (lambda *args: None, list(dsk))
-        dsk[f"prepare_target-{token}"] = (
-            _prepare_target,
-            f"checkpoint_0-{token}",
-            self.prepare_target,
-        )
-
-        # Store Chunk --------------------------------------------------------
-        keys = []
-        for i, chunk_key in enumerate(self.iter_chunks()):
-            k = (f"store_chunk-{token}", i)
-            dsk[k] = (_store_chunk, f"prepare_target-{token}", self.store_chunk, chunk_key)
-            keys.append(k)
-
-        # Finalize Target -----------------------------------------------------
-        dsk[f"checkpoint_1-{token}"] = (lambda *args: None, keys)
-        key = f"finalize_target-{token}"
-        dsk[key] = (_finalize_target, f"checkpoint_1-{token}", self.finalize_target)
-
-        return Delayed(key, dsk)
-
-    def to_prefect(self):
-        """Compile the recipe to a Prefect.Flow object."""
-        from prefect import Flow, task, unmapped
-
-        # TODO: allow recipes to customize which stages to run
-        cache_input_task = task(self.cache_input, name="cache_input")
-        prepare_target_task = task(self.prepare_target, name="prepare_target")
-        store_chunk_task = task(self.store_chunk, name="store_chunk")
-        finalize_target_task = task(self.finalize_target, name="finalize_target")
-
-        with Flow("pangeo-forge-recipe") as flow:
-            cache_task = cache_input_task.map(input_key=list(self.iter_inputs()))
-            upstream_tasks = [cache_task]
-            prepare_task = prepare_target_task(upstream_tasks=upstream_tasks)
-            store_task = store_chunk_task.map(
-                chunk_key=list(self.iter_chunks()), upstream_tasks=[unmapped(prepare_task)],
-            )
-            _ = finalize_target_task(upstream_tasks=[store_task])
-
-        return flow
-
-    def __iter__(self):
-        # TODO: allow recipes to customize which stages to run
-        yield self.cache_input, self.iter_inputs()
-        yield self.prepare_target, []
-        yield self.store_chunk, self.iter_chunks()
-        yield self.finalize_target, []
+        from pangeo_forge_recipes.executors.dask import DaskPipelineExecutor
 
-    # https://stackoverflow.com/questions/59986413/achieving-multiple-inheritance-using-python-dataclasses
-    def __post_init__(self):
-        # just intercept the __post_init__ calls so they
-        # aren't relayed to `object`
-        pass
-
-
-def closure(func: Callable) -> Callable:
-    """Wrap a method to eliminate the self keyword from its signature."""
-
-    # tried using @functools.wraps, but could not get it to work right
-    def wrapped(*args, **kwargs):
-        self = args[0]
-        if len(args) > 1:
-            args = args[1:]
-        else:
-            args = ()
-        new_func = partial(func, self, *args, **kwargs)
-        new_func.__name__ = getattr(func, "__name__", None)
-        return new_func
-
-    return wrapped
-
-
-def _prepare_target(checkpoint, func):
-    return func()
+        return DaskPipelineExecutor.compile(self._compiler())
 
+    def to_prefect(self):
+        from pangeo_forge_recipes.executors.prefect import PrefectPipelineExecutor
 
-def _store_chunk(checkpoint, func, input_key):
-    return func(input_key)
+        return PrefectPipelineExecutor.compile(self._compiler())
 
 
-def _finalize_target(checkpoint, func):
-    return func()
+RecipeCompiler = Callable[[BaseRecipe], Pipeline]
 
 
 @dataclass
-class FilePatternRecipeMixin:
+class FilePatternMixin:
     file_pattern: FilePattern
 
     def copy_pruned(self, nkeep: int = 2):
diff --git a/pangeo_forge_recipes/recipes/reference_hdf_zarr.py b/pangeo_forge_recipes/recipes/reference_hdf_zarr.py
index c8842238..deda7e6c 100644
--- a/pangeo_forge_recipes/recipes/reference_hdf_zarr.py
+++ b/pangeo_forge_recipes/recipes/reference_hdf_zarr.py
@@ -1,17 +1,19 @@
-import functools
+from __future__ import annotations
+
 import json
 import os
 from dataclasses import dataclass, field
-from typing import Callable, Dict, Hashable, Iterable, Optional
+from typing import Hashable, Iterable, Optional
 
 import fsspec
 import yaml
 from fsspec_reference_maker.combine import MultiZarrToZarr
 
-from ..patterns import FilePattern, Index
+from ..executors.base import Pipeline, Stage
+from ..patterns import Index
 from ..reference import create_hdf5_reference, unstrip_protocol
 from ..storage import FSSpecTarget, MetadataTarget, file_opener
-from .base import BaseRecipe, FilePatternRecipeMixin
+from .base import BaseRecipe, FilePatternMixin
 
 ChunkKey = Index
 
@@ -21,8 +23,83 @@ def no_op(*_, **__) -> None:
     return None
 
 
+def scan_file(chunk_key: ChunkKey, config: HDFReferenceRecipe):
+    assert config.metadata_cache is not None, "metadata_cache is required"
+    fname = config.file_pattern[chunk_key]
+    ref_fname = os.path.basename(fname + ".json")
+    with file_opener(fname, **config.netcdf_storage_options) as fp:
+        protocol = getattr(getattr(fp, "fs", None), "protocol", None)  # make mypy happy
+        target_url = unstrip_protocol(fname, protocol)
+        config.metadata_cache[ref_fname] = create_hdf5_reference(fp, target_url, fname)
+
+
+def finalize(config: HDFReferenceRecipe):
+    assert config.target is not None, "target is required"
+    assert config.metadata_cache is not None, "metadata_cache is required"
+    remote_protocol = fsspec.utils.get_protocol(next(config.file_pattern.items())[1])
+    concat_args = config.xarray_concat_args.copy()
+    if "dim" in concat_args:
+        raise ValueError(
+            "Please do not set 'dim' in xarray_concat_args. "
+            "It is determined automatically from the File Pattern."
+        )
+    concat_args["dim"] = config.file_pattern.concat_dims[0]  # there should only be one
+
+    files = list(
+        config.metadata_cache.getitems(list(config.metadata_cache.get_mapper())).values()
+    )  # returns dicts from remote
+    if len(files) == 1:
+        out = files[0]
+    else:
+        mzz = MultiZarrToZarr(
+            files,
+            remote_protocol=remote_protocol,
+            remote_options=config.netcdf_storage_options,
+            xarray_open_kwargs=config.xarray_open_kwargs,
+            xarray_concat_args=concat_args,
+        )
+        # mzz does not support directly writing to remote yet
+        # get dict version and push it
+        out = mzz.translate(None, template_count=config.template_count)
+    fs = config.target.fs
+    protocol = fs.protocol if isinstance(fs.protocol, str) else fs.protocol[0]
+    with config.target.open(config.output_json_fname, mode="wt") as f:
+        f.write(json.dumps(out))
+    spec = {
+        "sources": {
+            "data": {
+                "driver": "intake_xarray.xzarr.ZarrSource",
+                "description": "",  # could derive from data attrs or recipe
+                "args": {
+                    "urlpath": "reference://",
+                    "storage_options": {
+                        "fo": config.target._full_path(config.output_json_fname),
+                        "target_protocol": protocol,
+                        "target_options": config.output_storage_options,
+                        "remote_protocol": remote_protocol,
+                        "remote_options": config.netcdf_storage_options,
+                        "skip_instance_cache": True,
+                    },
+                    "chunks": {},  # can optimize access here
+                    "consolidated": False,
+                },
+            }
+        }
+    }
+    with config.target.open(config.output_intake_yaml_fname, mode="wt") as f:
+        yaml.dump(spec, f, default_flow_style=False)
+
+
+def hdf_reference_recipe_compiler(recipe: HDFReferenceRecipe) -> Pipeline:
+    stages = [
+        Stage(name="scan_file", function=scan_file, mappable=list(recipe.iter_inputs())),
+        Stage(name="finalize", function=finalize),
+    ]
+    return Pipeline(stages=stages, config=recipe)
+
+
 @dataclass
-class HDFReferenceRecipe(BaseRecipe, FilePatternRecipeMixin):
+class HDFReferenceRecipe(BaseRecipe, FilePatternMixin):
     """
     Generates reference files for each input netCDF, then combines
     into one ensemble output
@@ -58,6 +135,8 @@ class HDFReferenceRecipe(BaseRecipe, FilePatternRecipeMixin):
         Only used if `file_pattern` has more than one file.
     """
 
+    _compiler = hdf_reference_recipe_compiler
+
     # TODO: support chunked ("tree") aggregation: would entail processing each file
     #  in one stage, running a set of merges in a second step and doing
     #  a master merge in finalise. This would maybe map to iter_chunk,
@@ -86,130 +165,4 @@ def _validate_file_pattern(self):
             raise NotImplementedError("This Recipe class can't handle more than one concat dim.")
 
     def iter_inputs(self) -> Iterable[Hashable]:
-        return ()
-
-    @property
-    def cache_input(self) -> Callable[[Hashable], None]:
-        return no_op
-
-    @property
-    def prepare_target(self) -> Callable[[], None]:
-        """Prepare the recipe for execution by initializing the target.
-        Attribute that returns a callable function.
-        """
-        return no_op
-
-    def iter_chunks(self) -> Iterable[Hashable]:
-        """Iterate over all target chunks."""
-        for input_key in self.file_pattern:
-            yield input_key
-
-    @property
-    def store_chunk(self) -> Callable[[Hashable], None]:
-        """Store a chunk of data in the target.
-        """
-        return functools.partial(
-            _one_chunk,
-            file_pattern=self.file_pattern,
-            netcdf_storage_options=self.netcdf_storage_options,
-            metadata_cache=self.metadata_cache,
-        )
-
-    @property
-    def finalize_target(self) -> Callable[[], None]:
-        """Final step to finish the recipe after data has been written.
-        Attribute that returns a callable function.
-        """
-        proto = fsspec.utils.get_protocol(next(self.file_pattern.items())[1])
-        concat_args = self.xarray_concat_args.copy()
-        if "dim" in concat_args:
-            raise ValueError(
-                "Please do not set 'dim' in xarray_concat_args. "
-                "It is determined automatically from the File Pattern."
-            )
-        concat_args["dim"] = self.file_pattern.concat_dims[0]  # there should only be one
-        return functools.partial(
-            _finalize,
-            output_json_fname=self.output_json_fname,
-            output_intake_yaml_fname=self.output_intake_yaml_fname,
-            out_target=self.target,
-            metadata_cache=self.metadata_cache,
-            remote_protocol=proto,
-            output_storage_options=self.output_storage_options,
-            remote_options=self.netcdf_storage_options,
-            xr_open_kwargs=self.xarray_open_kwargs,
-            xr_concat_kwargs=concat_args,
-            template_count=self.template_count,
-        )
-
-
-def _one_chunk(
-    chunk_key: ChunkKey,
-    file_pattern: FilePattern,
-    netcdf_storage_options: Dict,
-    metadata_cache: MetadataTarget,
-):
-    fname = file_pattern[chunk_key]
-    ref_fname = os.path.basename(fname + ".json")
-    with file_opener(fname, **netcdf_storage_options) as fp:
-        protocol = getattr(getattr(fp, "fs", None), "protocol", None)  # make mypy happy
-        target_url = unstrip_protocol(fname, protocol)
-        metadata_cache[ref_fname] = create_hdf5_reference(fp, target_url, fname)
-
-
-def _finalize(
-    output_json_fname,
-    output_intake_yaml_fname,
-    out_target,
-    metadata_cache,
-    remote_protocol,
-    output_storage_options,
-    remote_options,
-    xr_open_kwargs,
-    xr_concat_kwargs,
-    template_count,
-):
-
-    files = list(
-        metadata_cache.getitems(list(metadata_cache.get_mapper())).values()
-    )  # returns dicts from remote
-    if len(files) == 1:
-        out = files[0]
-    else:
-        mzz = MultiZarrToZarr(
-            files,
-            remote_protocol=remote_protocol,
-            remote_options=remote_options,
-            xarray_open_kwargs=xr_open_kwargs,
-            xarray_concat_args=xr_concat_kwargs,
-        )
-        # mzz does not support directly writing to remote yet
-        # get dict version and push it
-        out = mzz.translate(None, template_count=template_count)
-    fs = out_target.fs
-    protocol = fs.protocol if isinstance(fs.protocol, str) else fs.protocol[0]
-    with out_target.open(output_json_fname, mode="wt") as f:
-        f.write(json.dumps(out))
-    spec = {
-        "sources": {
-            "data": {
-                "driver": "intake_xarray.xzarr.ZarrSource",
-                "description": "",  # could derive from data attrs or recipe
-                "args": {
-                    "urlpath": "reference://",
-                    "storage_options": {
-                        "fo": out_target._full_path(output_json_fname),
-                        "target_protocol": protocol,
-                        "target_options": output_storage_options,
-                        "remote_protocol": remote_protocol,
-                        "remote_options": remote_options,
-                        "skip_instance_cache": True,
-                    },
-                    "chunks": {},  # can optimize access here
-                    "consolidated": False,
-                },
-            }
-        }
-    }
-    with out_target.open(output_intake_yaml_fname, mode="wt") as f:
-        yaml.dump(spec, f, default_flow_style=False)
+        yield from self.file_pattern
diff --git a/pangeo_forge_recipes/recipes/xarray_zarr.py b/pangeo_forge_recipes/recipes/xarray_zarr.py
index d81e58af..c7bc036d 100644
--- a/pangeo_forge_recipes/recipes/xarray_zarr.py
+++ b/pangeo_forge_recipes/recipes/xarray_zarr.py
@@ -1,8 +1,8 @@
 """
 A Pangeo Forge Recipe
 """
+from __future__ import annotations
 
-import functools
 import itertools
 import logging
 import os
@@ -11,7 +11,7 @@
 from dataclasses import dataclass, field, replace
 from itertools import chain, product
 from math import ceil
-from typing import Callable, Dict, Hashable, Iterator, List, Optional, Sequence, Set, Tuple
+from typing import Callable, Dict, Iterator, List, Optional, Sequence, Set, Tuple
 
 import dask
 import fsspec
@@ -20,11 +20,12 @@
 import zarr
 
 from ..chunk_grid import ChunkGrid
+from ..executors.base import Pipeline, Stage
 from ..patterns import CombineOp, DimIndex, FilePattern, Index
 from ..reference import create_hdf5_reference, unstrip_protocol
-from ..storage import AbstractTarget, CacheFSSpecTarget, MetadataTarget, file_opener
+from ..storage import CacheFSSpecTarget, FSSpecTarget, MetadataTarget, file_opener
 from ..utils import calc_subsets, fix_scalar_attr_encoding, lock_for_conflicts
-from .base import BaseRecipe, FilePatternRecipeMixin
+from .base import BaseRecipe, FilePatternMixin
 
 # use this filename to store global recipe metadata in the metadata_cache
 # it will be written once (by prepare_target) and read many times (by store_chunk)
@@ -90,7 +91,7 @@ def inputs_for_chunk(
     return input_keys
 
 
-def expand_target_dim(target: CacheFSSpecTarget, concat_dim: Optional[str], dimsize: int) -> None:
+def expand_target_dim(target: FSSpecTarget, concat_dim: Optional[str], dimsize: int) -> None:
     target_mapper = target.get_mapper()
     zgroup = zarr.open_group(target_mapper)
     ds = open_target(target)
@@ -111,7 +112,7 @@ def expand_target_dim(target: CacheFSSpecTarget, concat_dim: Optional[str], dims
         zgroup[concat_dim][:] = 0
 
 
-def open_target(target: CacheFSSpecTarget) -> xr.Dataset:
+def open_target(target: FSSpecTarget) -> xr.Dataset:
     return xr.open_zarr(target.get_mapper())
 
 
@@ -144,216 +145,152 @@ def chunk_position(chunk_key: ChunkKey) -> int:
     return subset_factor * concat_idx + subset_idx
 
 
-def cache_input_metadata(
-    input_key: InputKey,
-    metadata_cache: Optional[MetadataTarget],
-    file_pattern: FilePattern,
-    input_cache: Optional[CacheFSSpecTarget],
-    cache_inputs: bool,
-    copy_input_to_local_file: bool,
-    xarray_open_kwargs: dict,
-    delete_input_encoding: bool,
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
-    open_input_with_fsspec_reference: bool,
-) -> None:
-    if metadata_cache is None:
-        raise ValueError("metadata_cache is not set.")
-    logger.info(f"Caching metadata for input '{input_key!s}'")
-    with open_input(
-        input_key,
-        file_pattern=file_pattern,
-        input_cache=input_cache,
-        cache_inputs=cache_inputs,
-        copy_input_to_local_file=copy_input_to_local_file,
-        xarray_open_kwargs=xarray_open_kwargs,
-        delete_input_encoding=delete_input_encoding,
-        process_input=process_input,
-        open_input_with_fsspec_reference=open_input_with_fsspec_reference,
-        metadata_cache=metadata_cache,
-    ) as ds:
-        input_metadata = ds.to_dict(data=False)
-        metadata_cache[_input_metadata_fname(input_key)] = input_metadata
-
-
-def make_input_reference(
-    input_key: InputKey,
-    metadata_cache: MetadataTarget,
-    file_pattern: FilePattern,
-    input_cache: Optional[CacheFSSpecTarget],
-    copy_input_to_local_file: bool,
-) -> None:
-
-    fname = file_pattern[input_key]
-    if input_cache is None:
-        protocol = fsspec.utils.get_protocol(fname)
-        url = unstrip_protocol(fname, protocol)
-    else:
-        url = unstrip_protocol(input_cache._full_path(fname), input_cache.fs.protocol)
-    with file_opener(
-        fname,
-        cache=input_cache,
-        copy_to_local=copy_input_to_local_file,
-        bypass_open=False,
-        secrets=file_pattern.query_string_secrets,
-        **file_pattern.fsspec_open_kwargs,
-    ) as fp:
-        ref_data = create_hdf5_reference(fp, url, fname)
-
-    ref_fname = _input_reference_fname(input_key)
-    metadata_cache[ref_fname] = ref_data
-
-
-def cache_input(
-    input_key: InputKey,
-    cache_inputs: bool,
-    input_cache: Optional[CacheFSSpecTarget],
-    file_pattern: FilePattern,
-    cache_metadata: bool,
-    copy_input_to_local_file: bool,
-    xarray_open_kwargs: dict,
-    delete_input_encoding: bool,
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
-    metadata_cache: Optional[MetadataTarget],
-    open_input_with_fsspec_reference: bool,
-) -> None:
-    if cache_inputs:
-        if file_pattern.is_opendap:
+def cache_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> None:
+    if config.cache_inputs:
+        if config.file_pattern.is_opendap:
             raise ValueError("Can't cache opendap inputs")
-        if input_cache is None:
+        if config.input_cache is None:
             raise ValueError("input_cache is not set.")
         logger.info(f"Caching input '{input_key!s}'")
-        fname = file_pattern[input_key]
-        input_cache.cache_file(
-            fname, file_pattern.query_string_secrets, **file_pattern.fsspec_open_kwargs
+        fname = config.file_pattern[input_key]
+        config.input_cache.cache_file(
+            fname,
+            config.file_pattern.query_string_secrets,
+            **config.file_pattern.fsspec_open_kwargs,
         )
 
-    if open_input_with_fsspec_reference:
-        cache = input_cache if cache_inputs else None
-        if file_pattern.is_opendap:
+    if config.cache_metadata:
+        if config.metadata_cache is None:
+            raise ValueError("metadata_cache is not set.")
+        logger.info(f"Caching metadata for input '{input_key!s}'")
+        with open_input(input_key, config=config) as ds:
+            input_metadata = ds.to_dict(data=False)
+            config.metadata_cache[_input_metadata_fname(input_key)] = input_metadata
+
+    if config.open_input_with_fsspec_reference:
+        if config.file_pattern.is_opendap:
             raise ValueError("Can't make references for opendap inputs")
-        if metadata_cache is None:
+        if config.metadata_cache is None:
             raise ValueError("Can't make references; no metadata_cache assigned")
-        make_input_reference(
-            input_key, metadata_cache, file_pattern, cache, copy_input_to_local_file,
-        )
-
-    if cache_metadata:
-        cache_input_metadata(
-            input_key,
-            file_pattern=file_pattern,
-            input_cache=input_cache,
-            cache_inputs=cache_inputs,
-            copy_input_to_local_file=copy_input_to_local_file,
-            xarray_open_kwargs=xarray_open_kwargs,
-            delete_input_encoding=delete_input_encoding,
-            process_input=process_input,
-            metadata_cache=metadata_cache,
-            open_input_with_fsspec_reference=open_input_with_fsspec_reference,
-        )
+        fname = config.file_pattern[input_key]
+        if config.input_cache is None:
+            protocol = fsspec.utils.get_protocol(fname)
+            url = unstrip_protocol(fname, protocol)
+        else:
+            url = unstrip_protocol(
+                config.input_cache._full_path(fname), config.input_cache.fs.protocol
+            )
+        with file_opener(
+            fname,
+            cache=config.input_cache,
+            copy_to_local=config.copy_input_to_local_file,
+            bypass_open=False,
+            secrets=config.file_pattern.query_string_secrets,
+            **config.file_pattern.fsspec_open_kwargs,
+        ) as fp:
+            ref_data = create_hdf5_reference(fp, url, fname)
+        ref_fname = _input_reference_fname(input_key)
+        config.metadata_cache[ref_fname] = ref_data
 
 
 def region_and_conflicts_for_chunk(
-    chunk_key: ChunkKey,
-    inputs_per_chunk: int,
-    nitems_per_input: Optional[int],
-    file_pattern: FilePattern,
-    concat_dim_chunks: Optional[int],
-    concat_dim: str,
-    metadata_cache: Optional[MetadataTarget],
-    subset_inputs: Optional[SubsetSpec],
+    config: XarrayZarrRecipe, chunk_key: ChunkKey
 ) -> Tuple[Dict[str, slice], Dict[str, Set[int]]]:
     # return a dict suitable to pass to xr.to_zarr(region=...)
     # specifies where in the overall array to put this chunk's data
     # also return the conflicts with other chunks
 
-    if nitems_per_input:
-        input_sequence_lens = (nitems_per_input,) * file_pattern.dims[concat_dim]  # type: ignore
+    if config.nitems_per_input:
+        input_sequence_lens = (config.nitems_per_input,) * config.file_pattern.dims[
+            config.concat_dim
+        ]  # type: ignore
     else:
-        assert metadata_cache is not None  # for mypy
-        global_metadata = metadata_cache[_GLOBAL_METADATA_KEY]
+        assert config.metadata_cache is not None  # for mypy
+        global_metadata = config.metadata_cache[_GLOBAL_METADATA_KEY]
         input_sequence_lens = global_metadata["input_sequence_lens"]
     total_len = sum(input_sequence_lens)
 
     # for now this will just have one key since we only allow one concat_dim
     # but it could expand to accomodate multiple concat dims
-    chunk_index = {concat_dim: chunk_position(chunk_key)}
+    chunk_index = {config.concat_dim: chunk_position(chunk_key)}
 
-    input_chunk_grid = ChunkGrid({concat_dim: input_sequence_lens})
-    if subset_inputs and concat_dim in subset_inputs:
+    input_chunk_grid = ChunkGrid({config.concat_dim: input_sequence_lens})
+    if config.subset_inputs and config.concat_dim in config.subset_inputs:
         assert (
-            inputs_per_chunk == 1
+            config.inputs_per_chunk == 1
         ), "Doesn't make sense to have multiple inputs per chunk plus subsetting"
-        chunk_grid = input_chunk_grid.subset(subset_inputs)
-    elif inputs_per_chunk > 1:
-        chunk_grid = input_chunk_grid.consolidate({concat_dim: inputs_per_chunk})
+        chunk_grid = input_chunk_grid.subset(config.subset_inputs)
+    elif config.inputs_per_chunk > 1:
+        chunk_grid = input_chunk_grid.consolidate({config.concat_dim: config.inputs_per_chunk})
     else:
         chunk_grid = input_chunk_grid
-    assert chunk_grid.shape[concat_dim] == total_len
+    assert chunk_grid.shape[config.concat_dim] == total_len
 
     region = chunk_grid.chunk_index_to_array_slice(chunk_index)
 
-    assert concat_dim_chunks is not None
-    target_grid = ChunkGrid.from_uniform_grid({concat_dim: (concat_dim_chunks, total_len)})
+    assert config.concat_dim_chunks is not None
+    target_grid = ChunkGrid.from_uniform_grid(
+        {config.concat_dim: (config.concat_dim_chunks, total_len)}
+    )
     conflicts = chunk_grid.chunk_conflicts(chunk_index, target_grid)
 
     return region, conflicts
 
 
 @contextmanager
-def open_input(
-    input_key: InputKey,
-    file_pattern: FilePattern,
-    input_cache: Optional[CacheFSSpecTarget],
-    cache_inputs: bool,
-    copy_input_to_local_file: bool,
-    xarray_open_kwargs: dict,
-    delete_input_encoding: bool,
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
-    open_input_with_fsspec_reference: bool,
-    metadata_cache: Optional[MetadataTarget],
-) -> xr.Dataset:
-    fname = file_pattern[input_key]
+def open_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> xr.Dataset:
+    fname = config.file_pattern[input_key]
     logger.info(f"Opening input with Xarray {input_key!s}: '{fname}'")
 
-    if file_pattern.is_opendap:
-        if input_cache:
+    if config.file_pattern.is_opendap:
+        if config.input_cache:
             raise ValueError("Can't cache opendap inputs")
-        if copy_input_to_local_file:
+        if config.copy_input_to_local_file:
             raise ValueError("Can't copy opendap inputs to local file")
-        if open_input_with_fsspec_reference:
+        if config.open_input_with_fsspec_reference:
             raise ValueError("Can't open opendap inputs with fsspec-reference-maker")
 
-    if open_input_with_fsspec_reference:
-        if metadata_cache is None:
+    if config.open_input_with_fsspec_reference:
+        if config.metadata_cache is None:
             raise ValueError("metadata_cache is not set.")
         from fsspec.implementations.reference import ReferenceFileSystem
 
-        reference_data = metadata_cache[_input_reference_fname(input_key)]
+        reference_data = config.metadata_cache[_input_reference_fname(input_key)]
         # TODO: figure out how to set this for the cache target
-        remote_protocol = fsspec.utils.get_protocol(next(file_pattern.items())[1])
+        remote_protocol = fsspec.utils.get_protocol(next(config.file_pattern.items())[1])
         ref_fs = ReferenceFileSystem(
             reference_data, remote_protocol=remote_protocol, skip_instance_cache=True
         )
-
         mapper = ref_fs.get_mapper("/")
         # Doesn't really need to be a context manager, but that's how this function works
-        with xr.open_dataset(mapper, engine="zarr", chunks={}, consolidated=False) as ds:
-            yield ds
+        with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
+            with xr.open_dataset(mapper, engine="zarr", chunks={}, consolidated=False) as ds:
+                logger.debug("successfully opened reference dataset with zarr")
+
+                if config.delete_input_encoding:
+                    for var in ds.variables:
+                        ds[var].encoding = {}
+
+                if config.process_input is not None:
+                    ds = config.process_input(ds, str(fname))
+
+                logger.debug(f"{ds}")
+                yield ds
 
     else:
 
-        cache = input_cache if cache_inputs else None
+        cache = config.input_cache if config.cache_inputs else None
 
         with file_opener(
             fname,
             cache=cache,
-            copy_to_local=copy_input_to_local_file,
-            bypass_open=file_pattern.is_opendap,
-            secrets=file_pattern.query_string_secrets,
-            **file_pattern.fsspec_open_kwargs,
+            copy_to_local=config.copy_input_to_local_file,
+            bypass_open=config.file_pattern.is_opendap,
+            secrets=config.file_pattern.query_string_secrets,
+            **config.file_pattern.fsspec_open_kwargs,
         ) as f:
             with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                kw = xarray_open_kwargs.copy()
+                kw = config.xarray_open_kwargs.copy()
                 if "engine" not in kw:
                     kw["engine"] = "h5netcdf"
                 logger.debug(f"about to enter xr.open_dataset context on {f}")
@@ -361,12 +298,12 @@ def open_input(
                     logger.debug("successfully opened dataset")
                     ds = fix_scalar_attr_encoding(ds)
 
-                    if delete_input_encoding:
+                    if config.delete_input_encoding:
                         for var in ds.variables:
                             ds[var].encoding = {}
 
-                    if process_input is not None:
-                        ds = process_input(ds, str(fname))
+                    if config.process_input is not None:
+                        ds = config.process_input(ds, str(fname))
 
                     logger.debug(f"{ds}")
                     yield ds
@@ -386,46 +323,14 @@ def subset_dataset(ds: xr.Dataset, subset_spec: DimIndex) -> xr.Dataset:
 
 
 @contextmanager
-def open_chunk(
-    chunk_key: ChunkKey,
-    inputs_per_chunk: int,
-    concat_dim: Optional[str],
-    xarray_concat_kwargs: dict,
-    process_chunk: Optional[Callable[[xr.Dataset], xr.Dataset]],
-    target_chunks: Dict[str, int],
-    file_pattern: FilePattern,
-    input_cache: Optional[CacheFSSpecTarget],
-    cache_inputs: bool,
-    copy_input_to_local_file: bool,
-    xarray_open_kwargs: dict,
-    delete_input_encoding: bool,
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
-    open_input_with_fsspec_reference: bool,
-    metadata_cache: Optional[MetadataTarget],
-) -> xr.Dataset:
+def open_chunk(chunk_key: ChunkKey, *, config: XarrayZarrRecipe) -> xr.Dataset:
     logger.info(f"Opening inputs for chunk {chunk_key!s}")
-    ninputs = file_pattern.dims[file_pattern.concat_dims[0]]
-    inputs = inputs_for_chunk(chunk_key, inputs_per_chunk, ninputs)
+    ninputs = config.file_pattern.dims[config.file_pattern.concat_dims[0]]
+    inputs = inputs_for_chunk(chunk_key, config.inputs_per_chunk, ninputs)
 
     # need to open an unknown number of contexts at the same time
     with ExitStack() as stack:
-        dsets = [
-            stack.enter_context(
-                open_input(
-                    i,
-                    file_pattern=file_pattern,
-                    input_cache=input_cache,
-                    cache_inputs=cache_inputs,
-                    copy_input_to_local_file=copy_input_to_local_file,
-                    xarray_open_kwargs=xarray_open_kwargs,
-                    delete_input_encoding=delete_input_encoding,
-                    process_input=process_input,
-                    open_input_with_fsspec_reference=open_input_with_fsspec_reference,
-                    metadata_cache=metadata_cache,
-                )
-            )
-            for i in inputs
-        ]
+        dsets = [stack.enter_context(open_input(input_key, config=config)) for input_key in inputs]
 
         # subset before chunking; hopefully lazy
         subset_dims = [dim_idx for dim_idx in chunk_key if dim_idx.operation == CombineOp.SUBSET]
@@ -442,23 +347,23 @@ def open_chunk(
             # During concat, attributes and encoding are taken from the first dataset
             # https://github.com/pydata/xarray/issues/1614
             with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                ds = xr.concat(dsets, concat_dim, **xarray_concat_kwargs)
+                ds = xr.concat(dsets, config.concat_dim, **config.xarray_concat_kwargs)
         elif len(dsets) == 1:
             ds = dsets[0]
         else:  # pragma: no cover
             assert False, "Should never happen"
 
-        if process_chunk is not None:
+        if config.process_chunk is not None:
             with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                ds = process_chunk(ds)
+                ds = config.process_chunk(ds)
 
         with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
             logger.debug(f"{ds}")
 
-        if target_chunks:
+        if config.target_chunks:
             # The input may be too large to process in memory at once, so
             # rechunk it to the target chunks.
-            ds = ds.chunk(target_chunks)
+            ds = ds.chunk(config.target_chunks)
         yield ds
 
 
@@ -479,6 +384,7 @@ def calculate_sequence_lens(
     concat_dim = file_pattern.concat_dims[0]
 
     if nitems_per_input:
+        concat_dim = file_pattern.concat_dims[0]
         return list((nitems_per_input,) * file_pattern.dims[concat_dim])
 
     # read per-input metadata; this is distinct from global metadata
@@ -499,32 +405,15 @@ def calculate_sequence_lens(
     return np.atleast_1d(sequence_lens.squeeze()).tolist()
 
 
-def prepare_target(
-    target: CacheFSSpecTarget,
-    target_chunks: Dict[str, int],
-    init_chunks: List[ChunkKey],
-    concat_dim: Optional[str],
-    nitems_per_input: Optional[int],
-    file_pattern: FilePattern,
-    inputs_per_chunk: int,
-    cache_metadata: bool,
-    xarray_concat_kwargs: dict,
-    process_chunk: Optional[Callable[[xr.Dataset], xr.Dataset]],
-    input_cache: Optional[CacheFSSpecTarget],
-    cache_inputs: bool,
-    copy_input_to_local_file: bool,
-    xarray_open_kwargs: dict,
-    delete_input_encoding: bool,
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
-    metadata_cache: Optional[MetadataTarget],
-    open_input_with_fsspec_reference: bool,
-) -> None:
+def prepare_target(*, config: XarrayZarrRecipe) -> None:
+    if config.target is None:
+        raise ValueError("Cannot proceed without a target")
     try:
-        ds = open_target(target)
+        ds = open_target(config.target)
         logger.info("Found an existing dataset in target")
         logger.debug(f"{ds}")
 
-        if target_chunks:
+        if config.target_chunks:
             # TODO: check that target_chunks id compatibile with the
             # existing chunks
             pass
@@ -532,33 +421,26 @@ def prepare_target(
         logger.info("Creating a new dataset in target")
 
         # need to rewrite this as an append loop
+
+        def filter_init_chunks(chunk_key):
+            for dim_idx in chunk_key:
+                if (dim_idx.operation != CombineOp.MERGE) and (dim_idx.index > 0):
+                    return False
+            return True
+
+        init_chunks = list(filter(filter_init_chunks, config.iter_chunks()))
+
         for chunk_key in init_chunks:
-            with open_chunk(
-                chunk_key=chunk_key,
-                inputs_per_chunk=inputs_per_chunk,
-                concat_dim=concat_dim,
-                xarray_concat_kwargs=xarray_concat_kwargs,
-                process_chunk=process_chunk,
-                target_chunks=target_chunks,
-                file_pattern=file_pattern,
-                input_cache=input_cache,
-                cache_inputs=cache_inputs,
-                copy_input_to_local_file=copy_input_to_local_file,
-                xarray_open_kwargs=xarray_open_kwargs,
-                delete_input_encoding=delete_input_encoding,
-                process_input=process_input,
-                open_input_with_fsspec_reference=open_input_with_fsspec_reference,
-                metadata_cache=metadata_cache,
-            ) as ds:
+            with open_chunk(chunk_key, config=config) as ds:
                 # ds is already chunked
 
                 # https://github.com/pydata/xarray/blob/5287c7b2546fc8848f539bb5ee66bb8d91d8496f/xarray/core/variable.py#L1069
                 for v in ds.variables:
-                    if target_chunks:
+                    if config.target_chunks:
                         this_var = ds[v]
                         chunks = {
                             this_var.get_axis_num(dim): chunk
-                            for dim, chunk in target_chunks.items()
+                            for dim, chunk in config.target_chunks.items()
                             if dim in this_var.dims
                         }
                         encoding_chunks = tuple(
@@ -574,11 +456,11 @@ def prepare_target(
                 # Variables that are loaded will be written even with compute=False
                 # TODO: make this behavior customizable
                 for v in ds.variables:
-                    if concat_dim not in ds[v].dims:
+                    if config.concat_dim not in ds[v].dims:
                         ds[v].load()
 
-                target_mapper = target.get_mapper()
-                logger.info(f"Storing dataset in {target.root_path}")
+                target_mapper = config.target.get_mapper()
+                logger.info(f"Storing dataset in {config.target.root_path}")
                 logger.debug(f"{ds}")
                 with warnings.catch_warnings():
                     warnings.simplefilter(
@@ -589,78 +471,32 @@ def prepare_target(
     # Regardless of whether there is an existing dataset or we are creating a new one,
     # we need to expand the concat_dim to hold the entire expected size of the data
     input_sequence_lens = calculate_sequence_lens(
-        nitems_per_input, file_pattern, metadata_cache=metadata_cache,
+        config.nitems_per_input, config.file_pattern, config.metadata_cache,
     )
     n_sequence = sum(input_sequence_lens)
-    logger.info(f"Expanding target concat dim '{concat_dim}' to size {n_sequence}")
-    expand_target_dim(target, concat_dim, n_sequence)
+    logger.info(f"Expanding target concat dim '{config.concat_dim}' to size {n_sequence}")
+    expand_target_dim(config.target, config.concat_dim, n_sequence)
     # TODO: handle possible subsetting
     # The init chunks might not cover the whole dataset along multiple dimensions!
 
-    if cache_metadata:
+    if config.cache_metadata:
         # if nitems_per_input is not constant, we need to cache this info
-        assert metadata_cache is not None  # for mypy
+        assert config.metadata_cache is not None  # for mypy
         recipe_meta = {"input_sequence_lens": input_sequence_lens}
-        metadata_cache[_GLOBAL_METADATA_KEY] = recipe_meta
+        config.metadata_cache[_GLOBAL_METADATA_KEY] = recipe_meta
 
 
-def store_chunk(
-    chunk_key: ChunkKey,
-    target: CacheFSSpecTarget,
-    concat_dim: str,
-    nitems_per_input: Optional[int],
-    file_pattern: FilePattern,
-    inputs_per_chunk: int,
-    subset_inputs: SubsetSpec,
-    concat_dim_chunks: Optional[int],
-    lock_timeout: Optional[int],
-    xarray_concat_kwargs: dict,
-    process_chunk: Optional[Callable[[xr.Dataset], xr.Dataset]],
-    target_chunks: Dict[str, int],
-    input_cache: Optional[CacheFSSpecTarget],
-    cache_inputs: bool,
-    copy_input_to_local_file: bool,
-    xarray_open_kwargs: dict,
-    delete_input_encoding: bool,
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
-    metadata_cache: Optional[MetadataTarget],
-    open_input_with_fsspec_reference: bool,
-) -> None:
-    if target is None:
+def store_chunk(chunk_key: ChunkKey, *, config: XarrayZarrRecipe) -> None:
+    if config.target is None:
         raise ValueError("target has not been set.")
 
-    with open_chunk(
-        chunk_key=chunk_key,
-        inputs_per_chunk=inputs_per_chunk,
-        concat_dim=concat_dim,
-        xarray_concat_kwargs=xarray_concat_kwargs,
-        process_chunk=process_chunk,
-        target_chunks=target_chunks,
-        file_pattern=file_pattern,
-        input_cache=input_cache,
-        cache_inputs=cache_inputs,
-        copy_input_to_local_file=copy_input_to_local_file,
-        xarray_open_kwargs=xarray_open_kwargs,
-        delete_input_encoding=delete_input_encoding,
-        process_input=process_input,
-        open_input_with_fsspec_reference=open_input_with_fsspec_reference,
-        metadata_cache=metadata_cache,
-    ) as ds_chunk:
+    with open_chunk(chunk_key, config=config) as ds_chunk:
         # writing a region means that all the variables MUST have concat_dim
-        to_drop = [v for v in ds_chunk.variables if concat_dim not in ds_chunk[v].dims]
+        to_drop = [v for v in ds_chunk.variables if config.concat_dim not in ds_chunk[v].dims]
         ds_chunk = ds_chunk.drop_vars(to_drop)
 
-        target_mapper = target.get_mapper()
-        write_region, conflicts = region_and_conflicts_for_chunk(
-            chunk_key=chunk_key,
-            inputs_per_chunk=inputs_per_chunk,
-            nitems_per_input=nitems_per_input,
-            file_pattern=file_pattern,
-            concat_dim_chunks=concat_dim_chunks,
-            concat_dim=concat_dim,
-            metadata_cache=metadata_cache,
-            subset_inputs=subset_inputs,
-        )
+        target_mapper = config.target.get_mapper()
+        write_region, conflicts = region_and_conflicts_for_chunk(config, chunk_key)
 
         zgroup = zarr.open_group(target_mapper)
         for vname, var_coded in ds_chunk.variables.items():
@@ -679,12 +515,13 @@ def store_chunk(
                 )
                 if var.data.nbytes > MAX_MEMORY:
                     factor = round((var.data.nbytes / MAX_MEMORY), 2)
+                    cdim = config.concat_dim
                     logger.warning(
                         f"Variable {vname} of {var.data.nbytes} bytes is {factor} times larger "
                         f"than specified maximum variable array size of {MAX_MEMORY} bytes. "
-                        f'Consider re-instantiating recipe with `subset_inputs = {{"{concat_dim}": '
-                        f'{ceil(factor)}}}`. If `len(ds["{concat_dim}"])` < {ceil(factor)}, '
-                        f'substitute "{concat_dim}" for any name in ds["{vname}"].dims with length '
+                        f'Consider re-instantiating recipe with `subset_inputs = {{"{cdim}": '
+                        f'{ceil(factor)}}}`. If `len(ds["{cdim}"])` < {ceil(factor)}, '
+                        f'substitute "{cdim}" for any name in ds["{vname}"].dims with length '
                         f">= {ceil(factor)} or consider subsetting along multiple dimensions."
                         " Setting PANGEO_FORGE_MAX_MEMORY env variable changes the variable array"
                         " size which will trigger this warning."
@@ -699,7 +536,7 @@ def store_chunk(
                 for c in dim_conflicts
             ]
             logger.debug(f"Acquiring locks {lock_keys}")
-            with lock_for_conflicts(lock_keys, timeout=lock_timeout):
+            with lock_for_conflicts(lock_keys, timeout=config.lock_timeout):
                 logger.info(
                     f"Storing variable {vname} chunk {chunk_key!s} " f"to Zarr region {zarr_region}"
                 )
@@ -712,17 +549,13 @@ def _gather_coordinate_dimensions(group: zarr.Group) -> List[str]:
     )
 
 
-def finalize_target(
-    target: CacheFSSpecTarget,
-    consolidate_zarr: bool,
-    consolidate_dimension_coordinates: bool = True,
-) -> None:
-    if target is None:
+def finalize_target(*, config: XarrayZarrRecipe) -> None:
+    if config.target is None:
         raise ValueError("target has not been set.")
 
-    if consolidate_dimension_coordinates:
+    if config.consolidate_dimension_coordinates:
         logger.info("Consolidating dimension coordinate arrays")
-        target_mapper = target.get_mapper()
+        target_mapper = config.target.get_mapper()
         group = zarr.open(target_mapper, mode="a")
         # https://github.com/pangeo-forge/pangeo-forge-recipes/issues/214
         # intersect the dims from the array metadata with the Zarr group
@@ -744,25 +577,57 @@ def finalize_target(
             )
             new.attrs.update(attrs)
 
-    if consolidate_zarr:
+    if config.consolidate_zarr:
         logger.info("Consolidating Zarr metadata")
-        target_mapper = target.get_mapper()
+        target_mapper = config.target.get_mapper()
         zarr.consolidate_metadata(target_mapper)
 
 
+def xarray_zarr_recipe_compiler(recipe: XarrayZarrRecipe) -> Pipeline:
+    stages = [
+        Stage(name="cache_input", function=cache_input, mappable=list(recipe.iter_inputs())),
+        Stage(name="prepare_target", function=prepare_target),
+        Stage(name="store_chunk", function=store_chunk, mappable=list(recipe.iter_chunks())),
+        Stage(name="finalize_target", function=finalize_target),
+    ]
+    return Pipeline(stages=stages, config=recipe)
+
+
 # Notes about dataclasses:
 # - https://www.python.org/dev/peps/pep-0557/#inheritance
 # - https://stackoverflow.com/questions/51575931/class-inheritance-in-python-3-7-dataclasses
 
 
+_deprecation_message = (
+    "This method will be deprecated in v0.8.0. "
+    "Please call the equivalent function directly from the xarray_zarr module."
+)
+
+
 @dataclass
-class XarrayZarrRecipe(BaseRecipe, FilePatternRecipeMixin):
-    """This class represents a dataset composed of many individual NetCDF files.
+class XarrayZarrRecipe(BaseRecipe, FilePatternMixin):
+    """This configuration represents a dataset composed of many individual NetCDF files.
     This class uses Xarray to read and write data and writes its output to Zarr.
     The organization of the source files is described by the ``file_pattern``.
     Currently this recipe supports at most one ``MergeDim`` and one ``ConcatDim``
     in the File Pattern.
 
+    Manual Execution
+    ----------------
+
+    To manually execute a this recipe, run the following steps:
+
+    .. highlight:: python
+    .. code-block:: python
+
+        for input_key in recipe.iter_inputs():
+            recipe.cache_input(input_key)
+        recipe.prepare_target()
+        for chunk_key in recipe.iter_chunks():
+            recipe.store_chunk(chunk_key)
+        recipe.finalize_target()
+
+
     :param file_pattern: An object which describes the organization of the input files.
     :param inputs_per_chunk: The number of inputs to use in each chunk along the concat dim.
        Must be an integer >= 1.
@@ -803,9 +668,11 @@ class XarrayZarrRecipe(BaseRecipe, FilePatternRecipeMixin):
       the file with Xarray as a virtual Zarr dataset.
     """
 
+    _compiler = xarray_zarr_recipe_compiler
+
     inputs_per_chunk: int = 1
     target_chunks: Dict[str, int] = field(default_factory=dict)
-    target: Optional[AbstractTarget] = None
+    target: Optional[FSSpecTarget] = None
     input_cache: Optional[CacheFSSpecTarget] = None
     metadata_cache: Optional[MetadataTarget] = None
     cache_inputs: Optional[bool] = None
@@ -822,23 +689,30 @@ class XarrayZarrRecipe(BaseRecipe, FilePatternRecipeMixin):
     open_input_with_fsspec_reference: bool = False
 
     # internal attributes not meant to be seen or accessed by user
-    _concat_dim: str = field(default_factory=str, repr=False, init=False)
+    concat_dim: str = field(default_factory=str, repr=False, init=False)
     """The concatenation dimension name."""
 
-    _concat_dim_chunks: Optional[int] = field(default=None, repr=False, init=False)
+    concat_dim_chunks: Optional[int] = field(default=None, repr=False, init=False)
     """The desired chunking along the sequence dimension."""
 
-    _init_chunks: List[ChunkKey] = field(default_factory=list, repr=False, init=False)
+    init_chunks: List[ChunkKey] = field(default_factory=list, repr=False, init=False)
     """List of chunks needed to initialize the recipe."""
 
+    cache_metadata: bool = field(default=False, repr=False, init=False)
+    """Whether metadata caching is needed."""
+
+    nitems_per_input: Optional[int] = field(default=None, repr=False, init=False)
+    """How many items per input along concat_dim."""
+
     def __post_init__(self):
         self._validate_file_pattern()
+
         # from here on we know there is at most one merge dim and one concat dim
-        self._concat_dim = self.file_pattern.concat_dims[0]
-        self._cache_metadata = any(
+        self.concat_dim = self.file_pattern.concat_dims[0]
+        self.cache_metadata = any(
             [v is None for v in self.file_pattern.concat_sequence_lens.values()]
         )
-        self._nitems_per_input = self.file_pattern.nitems_per_input[self._concat_dim]
+        self.nitems_per_input = self.file_pattern.nitems_per_input[self.concat_dim]
 
         if self.file_pattern.is_opendap:
             if self.cache_inputs:
@@ -859,16 +733,19 @@ def __post_init__(self):
         elif self.cache_inputs is None:
             self.cache_inputs = True  # old defult
 
-        def filter_init_chunks(chunk_key):
-            for dim_idx in chunk_key:
-                if (dim_idx.operation != CombineOp.MERGE) and (dim_idx.index > 0):
-                    return False
-            return True
-
-        self._init_chunks = list(filter(filter_init_chunks, self.iter_chunks()))
-
         self._validate_input_and_chunk_keys()
-        self._set_target_chunks()
+
+        # set concat_dim_chunks
+        target_concat_dim_chunks = self.target_chunks.get(self.concat_dim)
+        if (self.nitems_per_input is None) and (target_concat_dim_chunks is None):
+            raise ValueError(
+                "Unable to determine target chunks. Please specify either "
+                "`target_chunks` or `nitems_per_input`"
+            )
+        elif target_concat_dim_chunks:
+            self.concat_dim_chunks = target_concat_dim_chunks
+        else:
+            self.concat_dim_chunks = self.nitems_per_input * self.inputs_per_chunk
 
     def _validate_file_pattern(self):
         if len(self.file_pattern.merge_dims) > 1:
@@ -895,104 +772,8 @@ def _validate_input_and_chunk_keys(self):
             print("Inputs_for_chunk", inputs_for_chunk(chunk_key, self.inputs_per_chunk, ninputs))
             raise ValueError("Inputs and chunks are inconsistent")
 
-    # below here are methods that are part of recipe execution
-    def _set_target_chunks(self):
-        target_concat_dim_chunks = self.target_chunks.get(self._concat_dim)
-        if (self._nitems_per_input is None) and (target_concat_dim_chunks is None):
-            raise ValueError(
-                "Unable to determine target chunks. Please specify either "
-                "`target_chunks` or `nitems_per_input`"
-            )
-        elif target_concat_dim_chunks:
-            self._concat_dim_chunks = target_concat_dim_chunks
-        else:
-            self._concat_dim_chunks = self._nitems_per_input * self.inputs_per_chunk
-
-    # Each stage of the recipe follows the same pattern:
-    # 1. A top-level function, e.g. `prepare_target`, that does the actual work.
-    # 2. A public property, e.g. `.prepare_target`, that calls the partially applied function
-    #    with the provided arguments if any (e.g. a chunk_key)
-    # This ensures that the actual function objects shipped to and executed on
-    # workers do not contain any references to the `recipe` object itself, which is complicated
-    # to serialize.
-
-    @property
-    def prepare_target(self) -> Callable[[], None]:
-        return functools.partial(
-            prepare_target,
-            target=self.target,
-            target_chunks=self.target_chunks,
-            init_chunks=self._init_chunks,
-            concat_dim=self._concat_dim,
-            nitems_per_input=self._nitems_per_input,
-            file_pattern=self.file_pattern,
-            inputs_per_chunk=self.inputs_per_chunk,
-            cache_metadata=self._cache_metadata,
-            xarray_concat_kwargs=self.xarray_concat_kwargs,
-            process_chunk=self.process_chunk,
-            input_cache=self.input_cache,
-            cache_inputs=self.cache_inputs,
-            copy_input_to_local_file=self.copy_input_to_local_file,
-            xarray_open_kwargs=self.xarray_open_kwargs,
-            delete_input_encoding=self.delete_input_encoding,
-            process_input=self.process_input,
-            metadata_cache=self.metadata_cache,
-            open_input_with_fsspec_reference=self.open_input_with_fsspec_reference,
-        )
-
-    @property
-    def cache_input(self) -> Callable[[Hashable], None]:
-        return functools.partial(
-            cache_input,
-            cache_inputs=self.cache_inputs,
-            input_cache=self.input_cache,
-            file_pattern=self.file_pattern,
-            cache_metadata=self._cache_metadata,
-            copy_input_to_local_file=self.copy_input_to_local_file,
-            xarray_open_kwargs=self.xarray_open_kwargs,
-            delete_input_encoding=self.delete_input_encoding,
-            process_input=self.process_input,
-            metadata_cache=self.metadata_cache,
-            open_input_with_fsspec_reference=self.open_input_with_fsspec_reference,
-        )
-
-    @property
-    def store_chunk(self) -> Callable[[Hashable], None]:
-        return functools.partial(
-            store_chunk,
-            target=self.target,
-            concat_dim=self._concat_dim,
-            nitems_per_input=self._nitems_per_input,
-            file_pattern=self.file_pattern,
-            inputs_per_chunk=self.inputs_per_chunk,
-            subset_inputs=self.subset_inputs,
-            concat_dim_chunks=self._concat_dim_chunks,
-            lock_timeout=self.lock_timeout,
-            xarray_concat_kwargs=self.xarray_concat_kwargs,
-            xarray_open_kwargs=self.xarray_open_kwargs,
-            process_chunk=self.process_chunk,
-            target_chunks=self.target_chunks,
-            input_cache=self.input_cache,
-            cache_inputs=self.cache_inputs,
-            copy_input_to_local_file=self.copy_input_to_local_file,
-            delete_input_encoding=self.delete_input_encoding,
-            process_input=self.process_input,
-            metadata_cache=self.metadata_cache,
-            open_input_with_fsspec_reference=self.open_input_with_fsspec_reference,
-        )
-
-    @property
-    def finalize_target(self) -> Callable[[], None]:
-        return functools.partial(
-            finalize_target,
-            target=self.target,
-            consolidate_zarr=self.consolidate_zarr,
-            consolidate_dimension_coordinates=self.consolidate_dimension_coordinates,
-        )
-
     def iter_inputs(self) -> Iterator[InputKey]:
-        for input_key in self.file_pattern:
-            yield input_key
+        yield from self.file_pattern
 
     def iter_chunks(self) -> Iterator[ChunkKey]:
         for input_key in self.iter_inputs():
@@ -1026,46 +807,60 @@ def iter_chunks(self) -> Iterator[ChunkKey]:
                 ]
                 yield Index((chunk_key_base + subset_dims))
 
+    # Below lie convience methods that help users develop and debug the recipe
+    # They will all be deprecated
+
     def inputs_for_chunk(self, chunk_key: ChunkKey) -> Sequence[InputKey]:
-        """Convenience function for users to introspect recipe."""
+        """Show which inputs are needed for which chunk.
+
+        :param chunk_key: Which chunk to examine
+        """
+        warnings.warn(_deprecation_message, DeprecationWarning)
         ninputs = self.file_pattern.dims[self.file_pattern.concat_dims[0]]
         return inputs_for_chunk(chunk_key, self.inputs_per_chunk, ninputs)
 
-    # ------------------------------------------------------------------------
-    # Convenience methods
     @contextmanager
-    def open_input(self, input_key):
-        with open_input(
-            input_key,
-            file_pattern=self.file_pattern,
-            input_cache=self.input_cache,
-            cache_inputs=self.cache_inputs,
-            copy_input_to_local_file=self.copy_input_to_local_file,
-            xarray_open_kwargs=self.xarray_open_kwargs,
-            delete_input_encoding=self.delete_input_encoding,
-            process_input=self.process_input,
-            open_input_with_fsspec_reference=self.open_input_with_fsspec_reference,
-            metadata_cache=self.metadata_cache,
-        ) as ds:
+    def open_input(self, input_key: InputKey) -> xr.Dataset:
+        """Open an input
+
+        :param chunk_key: Which input to open
+        """
+        warnings.warn(_deprecation_message, DeprecationWarning)
+        with open_input(input_key, config=self) as ds:
             yield ds
 
+    def cache_input(self, input_key: InputKey) -> None:
+        """Cache an input
+
+        :param chunk_key: Which input to cache
+        """
+        warnings.warn(_deprecation_message, DeprecationWarning)
+        cache_input(input_key, config=self)
+
+    def prepare_target(self) -> None:
+        """Prepare target for writing."""
+        warnings.warn(_deprecation_message, DeprecationWarning)
+        prepare_target(config=self)
+
     @contextmanager
-    def open_chunk(self, chunk_key):
-        with open_chunk(
-            chunk_key,
-            inputs_per_chunk=self.inputs_per_chunk,
-            concat_dim=self._concat_dim,
-            xarray_concat_kwargs=self.xarray_concat_kwargs,
-            process_chunk=self.process_chunk,
-            target_chunks=self.target_chunks,
-            file_pattern=self.file_pattern,
-            input_cache=self.input_cache,
-            cache_inputs=self.cache_input,
-            copy_input_to_local_file=self.copy_input_to_local_file,
-            xarray_open_kwargs=self.xarray_open_kwargs,
-            delete_input_encoding=self.delete_input_encoding,
-            process_input=self.process_input,
-            open_input_with_fsspec_reference=self.open_input_with_fsspec_reference,
-            metadata_cache=self.metadata_cache,
-        ) as ds:
+    def open_chunk(self, chunk_key: ChunkKey) -> xr.Dataset:
+        """Open the inputs that go into a chunk as a single dataset
+
+        :param chunk_key: Which chunk to open
+        """
+        warnings.warn(_deprecation_message, DeprecationWarning)
+        with open_chunk(chunk_key, config=self) as ds:
             yield ds
+
+    def store_chunk(self, chunk_key: ChunkKey) -> None:
+        """Store a chunk into the target
+
+        :param chunk_key: Which chunk to store
+        """
+        warnings.warn(_deprecation_message, DeprecationWarning)
+        store_chunk(chunk_key, config=self)
+
+    def finalize_target(self) -> None:
+        """Prepare target for writing."""
+        warnings.warn(_deprecation_message, DeprecationWarning)
+        finalize_target(config=self)
diff --git a/setup.cfg b/setup.cfg
index 058d340a..44b5ce30 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -60,7 +60,7 @@ max-line-length = 100
 
 [isort]
 known_first_party=pangeo_forge_recipes
-known_third_party=aiohttp,click,dask,fsspec,fsspec_reference_maker,numpy,pandas,prefect,pytest,pytest_lazyfixture,rechunker,setuptools,xarray,yaml,zarr
+known_third_party=aiohttp,click,dask,fsspec,fsspec_reference_maker,mypy_extensions,numpy,pandas,prefect,pytest,pytest_lazyfixture,setuptools,xarray,yaml,zarr
 multi_line_output=3
 include_trailing_comma=True
 force_grid_wrap=0
@@ -86,3 +86,5 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 [mypy-zarr.*]
 ignore_missing_imports = True
+[mypy-yaml.*]
+ignore_missing_imports = True


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index f7d0bb99..d3fb3f80 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -31,11 +31,6 @@
 # need to import this way (rather than use pytest.lazy_fixture) to make it work with dask
 from pytest_lazyfixture import lazy_fixture
 
-from pangeo_forge_recipes.executors import (
-    DaskPipelineExecutor,
-    PrefectPipelineExecutor,
-    PythonPipelineExecutor,
-)
 from pangeo_forge_recipes.patterns import (
     ConcatDim,
     FilePattern,
@@ -441,28 +436,6 @@ def redirect_logs():
     cluster.close()
 
 
-_executors = {
-    "python": PythonPipelineExecutor,
-    "dask": DaskPipelineExecutor,
-    "prefect": PrefectPipelineExecutor,
-    "prefect-dask": PrefectPipelineExecutor,
-}
-
-
-@pytest.fixture()
-def execute_recipe_manual():
-    def execute(r):
-        if r.cache_inputs:
-            for input_key in r.iter_inputs():
-                r.cache_input(input_key)
-        r.prepare_target()
-        for chunk_key in r.iter_chunks():
-            r.store_chunk(chunk_key)
-        r.finalize_target()
-
-    return execute
-
-
 @pytest.fixture()
 def execute_recipe_python():
     def execute(recipe):
@@ -503,11 +476,7 @@ def execute(recipe):
 
 
 @pytest.fixture(
-    params=[
-        lazy_fixture("execute_recipe_manual"),
-        lazy_fixture("execute_recipe_python"),
-        lazy_fixture("execute_recipe_dask"),
-    ],
+    params=[lazy_fixture("execute_recipe_python"), lazy_fixture("execute_recipe_dask")],
 )
 def execute_recipe_no_prefect(request):
     return request.param
@@ -521,11 +490,7 @@ def execute_recipe_with_prefect(request):
 
 
 @pytest.fixture(
-    params=[
-        lazy_fixture("execute_recipe_manual"),
-        lazy_fixture("execute_recipe_python"),
-        lazy_fixture("execute_recipe_prefect"),
-    ],
+    params=[lazy_fixture("execute_recipe_python"), lazy_fixture("execute_recipe_prefect")],
 )
 def execute_recipe_no_dask(request):
     return request.param
diff --git a/tests/recipe_tests/test_XarrayZarrRecipe.py b/tests/recipe_tests/test_XarrayZarrRecipe.py
index 80e44782..9b65f80d 100644
--- a/tests/recipe_tests/test_XarrayZarrRecipe.py
+++ b/tests/recipe_tests/test_XarrayZarrRecipe.py
@@ -2,7 +2,6 @@
 from dataclasses import replace
 from unittest.mock import patch
 
-import dask.core
 import pytest
 import xarray as xr
 import zarr
@@ -10,8 +9,7 @@
 # need to import this way (rather than use pytest.lazy_fixture) to make it work with dask
 from pytest_lazyfixture import lazy_fixture
 
-from pangeo_forge_recipes.patterns import ConcatDim, FilePattern
-from pangeo_forge_recipes.recipes.base import BaseRecipe
+from pangeo_forge_recipes.patterns import FilePattern
 from pangeo_forge_recipes.recipes.xarray_zarr import XarrayZarrRecipe
 
 
@@ -108,14 +106,6 @@ def netCDFtoZarr_subset_recipe(
 ]
 
 
-def test_to_pipelines_warns(netCDFtoZarr_recipe):
-    RecipeClass, file_pattern, kwargs, ds_expected, target = netCDFtoZarr_recipe
-
-    rec = RecipeClass(file_pattern, **kwargs)
-    with pytest.warns(FutureWarning):
-        rec.to_pipelines()
-
-
 @pytest.mark.parametrize("recipe_fixture", all_recipes)
 def test_recipe(recipe_fixture, execute_recipe):
     """The basic recipe test. Use this as a template for other tests."""
@@ -126,11 +116,23 @@ def test_recipe(recipe_fixture, execute_recipe):
     ds_actual = xr.open_zarr(target.get_mapper()).load()
     xr.testing.assert_identical(ds_actual, ds_expected)
 
-    with rec.open_input(next(rec.iter_inputs())):
-        pass
 
-    with rec.open_chunk(next(rec.iter_chunks())):
-        pass
+@pytest.mark.parametrize("recipe_fixture", all_recipes)
+def test_recipe_manual_execution(recipe_fixture):
+    """The basic recipe test. Use this as a template for other tests."""
+
+    RecipeClass, file_pattern, kwargs, ds_expected, target = recipe_fixture
+    rec = RecipeClass(file_pattern, **kwargs)
+
+    for input_key in rec.iter_inputs():
+        rec.cache_input(input_key)
+    rec.prepare_target()
+    for chunk_key in rec.iter_chunks():
+        rec.store_chunk(chunk_key)
+    rec.finalize_target()
+
+    ds_actual = xr.open_zarr(target.get_mapper()).load()
+    xr.testing.assert_identical(ds_actual, ds_expected)
 
 
 @pytest.mark.parametrize("recipe_fixture", all_recipes)
@@ -403,122 +405,3 @@ def test_lock_timeout(netCDFtoZarr_recipe_sequential_only, execute_recipe_no_das
         execute_recipe_no_dask(recipe)
 
     assert p.call_args[1]["timeout"] == 1
-
-
-class MyRecipe(BaseRecipe):
-    def __init__(self) -> None:
-        super().__init__()
-        self.cache = {}
-        self.target = None
-        self.finalized = False
-        self.cache_inputs = True
-
-    @property
-    def prepare_target(self):
-        def _():
-            self.target = {}
-
-        return _
-
-    @property
-    def cache_input(self):
-        def _(input_key):
-            self.cache[input_key] = input_key
-
-        return _
-
-    @property
-    def store_chunk(self):
-        def _(chunk_key):
-            self.target[chunk_key] = self.cache[chunk_key]
-
-        return _
-
-    @property
-    def finalize_target(self):
-        def _():
-            self.finalized = True
-
-        return _
-
-    def iter_inputs(self):
-        return iter(range(4))
-
-    def iter_chunks(self):
-        return iter(range(4))
-
-
-def test_base_recipe():
-    recipe = MyRecipe()
-    recipe.to_function()()
-    assert recipe.finalized
-    assert recipe.target == {i: i for i in range(4)}
-
-    import dask
-
-    dask.config.set(scheduler="single-threaded")
-    recipe = MyRecipe()
-    recipe.to_dask().compute()
-    assert recipe.finalized
-    assert recipe.target == {i: i for i in range(4)}
-
-    recipe = MyRecipe()
-    recipe.to_prefect().run()
-    assert recipe.finalized
-    assert recipe.target == {i: i for i in range(4)}
-
-
-def _make_filename_for_memory_usage_test(time):
-    import pandas as pd
-
-    input_url_pattern = (
-        "https://arthurhouhttps.pps.eosdis.nasa.gov/gpmdata/{yyyy}/{mm}/{dd}/"
-        "imerg/3B-HHR.MS.MRG.3IMERG.{yyyymmdd}-S{sh}{sm}00-E{eh}{em}59.{MMMM}.V06B.HDF5"
-    ).format(
-        yyyy=time.strftime("%Y"),
-        mm=time.strftime("%m"),
-        dd=time.strftime("%d"),
-        yyyymmdd=time.strftime("%Y%m%d"),
-        sh=time.strftime("%H"),
-        sm=time.strftime("%M"),
-        eh=time.strftime("%H"),
-        em=(time + pd.Timedelta("29 min")).strftime("%M"),
-        MMMM=f"{(time.hour*60 + time.minute):04}",
-    )
-    return input_url_pattern
-
-
-def _simple_func(*args, **kwargs):
-    return None
-
-
-@pytest.mark.timeout(90)
-@pytest.mark.filterwarnings("ignore:Large object")
-def test_memory_usage():
-    # https://github.com/pangeo-forge/pangeo-forge-recipes/issues/151
-    # Requires >4 GiB of memory to run.
-    pd = pytest.importorskip("pandas")
-    distributed = pytest.importorskip("distributed")
-
-    dates = pd.date_range("2020-05-31T00:00:00", "2021-05-31T23:59:59", freq="30min")
-    time_concat_dim = ConcatDim("time", dates, nitems_per_file=1)
-    pattern = FilePattern(_make_filename_for_memory_usage_test, time_concat_dim,)
-
-    recipe = XarrayZarrRecipe(
-        pattern, xarray_open_kwargs={"group": "Grid", "decode_coords": "all"}, inputs_per_chunk=1,
-    )
-
-    obj = recipe.to_dask()
-    dsk = obj.dask
-
-    for k, v in dsk.items():
-        if dask.core.istask(v):
-            _, *args = v
-            dsk[k] = (_simple_func,) + tuple(args)
-
-    with dask.config.set(
-        **{"distributed.worker.memory.pause": 0.95, "distributed.worker.memory.terminate": 0.9}
-    ):
-        with distributed.Client(n_workers=1, threads_per_worker=1, memory_limit="4G") as client:
-            print("submitting")
-            client.get(dsk, [obj.key])
diff --git a/tests/recipe_tests/test_graph_memory.py b/tests/recipe_tests/test_graph_memory.py
new file mode 100644
index 00000000..ffa7d4d1
--- /dev/null
+++ b/tests/recipe_tests/test_graph_memory.py
@@ -0,0 +1,72 @@
+import dask.core
+import pytest
+from dask.highlevelgraph import HighLevelGraph
+
+from pangeo_forge_recipes.patterns import ConcatDim, FilePattern
+from pangeo_forge_recipes.recipes.xarray_zarr import XarrayZarrRecipe
+
+
+def _make_filename_for_memory_usage_test(time):
+    import pandas as pd
+
+    input_url_pattern = (
+        "https://arthurhouhttps.pps.eosdis.nasa.gov/gpmdata/{yyyy}/{mm}/{dd}/"
+        "imerg/3B-HHR.MS.MRG.3IMERG.{yyyymmdd}-S{sh}{sm}00-E{eh}{em}59.{MMMM}.V06B.HDF5"
+    ).format(
+        yyyy=time.strftime("%Y"),
+        mm=time.strftime("%m"),
+        dd=time.strftime("%d"),
+        yyyymmdd=time.strftime("%Y%m%d"),
+        sh=time.strftime("%H"),
+        sm=time.strftime("%M"),
+        eh=time.strftime("%H"),
+        em=(time + pd.Timedelta("29 min")).strftime("%M"),
+        MMMM=f"{(time.hour*60 + time.minute):04}",
+    )
+    return input_url_pattern
+
+
+def _simple_func(*args, **kwargs):
+    return None
+
+
+@pytest.mark.timeout(90)
+@pytest.mark.filterwarnings("ignore:Large object")
+def test_memory_usage():
+    # https://github.com/pangeo-forge/pangeo-forge-recipes/issues/151
+    # Requires >4 GiB of memory to run.
+    pd = pytest.importorskip("pandas")
+    distributed = pytest.importorskip("distributed")
+
+    dates = pd.date_range("2020-05-31T00:00:00", "2021-05-31T23:59:59", freq="30min")
+    time_concat_dim = ConcatDim("time", dates, nitems_per_file=1)
+    pattern = FilePattern(_make_filename_for_memory_usage_test, time_concat_dim,)
+
+    recipe = XarrayZarrRecipe(
+        pattern, xarray_open_kwargs={"group": "Grid", "decode_coords": "all"}, inputs_per_chunk=1,
+    )
+
+    delayed = recipe.to_dask()
+
+    # in what follows, we rebuild the dask HighLevelGraph, replacing all of the
+    # actual functions with dummy functions that do nothin
+    dsk = delayed.dask
+    key = delayed.key
+    layers = dsk.layers
+    dependencies = dsk.dependencies
+    newlayers = {}
+    for layer_name, subgraph in layers.items():
+        new_subgraph = dict(subgraph)
+        for k, v in new_subgraph.items():
+            if dask.core.istask(v):
+                _, *args = v
+                new_subgraph[k] = (_simple_func,) + tuple(args)
+        newlayers[layer_name] = new_subgraph
+    new_hlg = HighLevelGraph(newlayers, dependencies)
+
+    with dask.config.set(
+        **{"distributed.worker.memory.pause": 0.95, "distributed.worker.memory.terminate": 0.9}
+    ):
+        with distributed.Client(n_workers=1, threads_per_worker=1, memory_limit="4G") as client:
+            print("submitting")
+            client.get(new_hlg, [key])
diff --git a/tests/test_locking.py b/tests/test_locking.py
index a5d42713..3f77268a 100644
--- a/tests/test_locking.py
+++ b/tests/test_locking.py
@@ -47,9 +47,9 @@ def do_stuff(n):
 @pytest.mark.parametrize(
     "shape, zarr_chunks, write_chunks",
     [
-        ((100,), (10,), (15,)),  # just a few conflicts
-        ((100,), (100,), (10,)),  # lots of conflicts!
-    ],
+        ((100,), (10,), (15,)),
+        ((100,), (100,), (10,)),
+    ],  # just a few conflicts  # lots of conflicts!
 )
 def test_locked_array_writing(shape, zarr_chunks, write_chunks, tmp_target, dask_cluster):
 
diff --git a/tests/test_pipelines.py b/tests/test_pipelines.py
new file mode 100644
index 00000000..59ab7206
--- /dev/null
+++ b/tests/test_pipelines.py
@@ -0,0 +1,72 @@
+"""
+Test Pipline Executors
+"""
+
+import pytest
+from pytest_lazyfixture import lazy_fixture
+
+from pangeo_forge_recipes.executors.base import Pipeline, Stage
+from pangeo_forge_recipes.executors.dask import DaskPipelineExecutor
+from pangeo_forge_recipes.executors.function import FunctionPipelineExecutor
+from pangeo_forge_recipes.executors.prefect import PrefectPipelineExecutor
+
+
+@pytest.fixture
+def pipeline_no_config(tmpdir_factory):
+    tmp = tmpdir_factory.mktemp("pipeline_data")
+
+    def func0(config=None):
+        tmp.join("func0.log").ensure(file=True)
+        assert not tmp.join("func1_a.log").check(file=True)
+
+    stage0 = Stage(function=func0, name="create_first_file")
+
+    def func1(arg, config=None):
+        tmp.join(f"func1_{arg}.log").ensure(file=True)
+
+    stage1 = Stage(function=func1, name="create_many_files", mappable=["a", "b", 3])
+    pipeline = Pipeline(stages=[stage0, stage1])
+    return pipeline, {}, tmp
+
+
+@pytest.fixture
+def pipeline_with_config(tmpdir_factory):
+    tmp = tmpdir_factory.mktemp("pipeline_data")
+
+    def func0(config=None):
+        prefix = config["prefix"]
+        tmp.join(f"{prefix}func0.log").ensure(file=True)
+        assert not tmp.join(f"{prefix}-func1_a.log").check(file=True)
+
+    stage0 = Stage(function=func0, name="create_first_file")
+
+    def func1(arg, config=None):
+        prefix = config["prefix"]
+        tmp.join(f"{prefix}func1_{arg}.log").ensure(file=True)
+
+    stage1 = Stage(function=func1, name="create_many_files", mappable=["a", "b", 3])
+    config = {"prefix": "special-"}
+    pipeline = Pipeline(stages=[stage0, stage1], config=config)
+    return pipeline, config, tmp
+
+
+@pytest.mark.parametrize(
+    "Executor", [FunctionPipelineExecutor, PrefectPipelineExecutor, DaskPipelineExecutor],
+)
+@pytest.mark.parametrize(
+    "pipeline_config_tmpdir",
+    [lazy_fixture("pipeline_no_config"), lazy_fixture("pipeline_with_config")],
+)
+def test_pipeline(pipeline_config_tmpdir, Executor):
+    pipeline, config, tmpdir = pipeline_config_tmpdir
+    plan = Executor.compile(pipeline)
+    Executor.execute(plan)
+
+    prefix = config.get("prefix", "")
+    for fname in [
+        f"{prefix}func0.log",
+        f"{prefix}func1_a.log",
+        f"{prefix}func1_b.log",
+        f"{prefix}func1_3.log",
+    ]:
+        assert tmpdir.join(fname).check(file=True)