prune old stuff
DELETE
pangeo_forge/pipelines/README.md
-# `pangeo_forge.pipelines`
-
-A collection of Pangeo-forge _Pipelines_.
-
-## Details
-
-- An abstract Pipeline class is defined in `base.py`.
-- A general file naming convention might be `{source}_{container}_{target}.py`. For example, a pipeline that downloads netCDF files from a remote http server and publishes a Zarr store would be `http_xarray_zarr.py`. The main class in each module would follow a similar convention, e.g. `HttpXarrayZarr`.

DELETE
pangeo_forge/pipelines/__init__.py
pangeo_forge/pipelines/__init__.py
DELETE
pangeo_forge/pipelines/base.py
-"""
-Base class for most pipelines.
-
-Design
-------
-
-To the extent possible, we want recipe maintainers to focus on the code
-needed to express the data transformations. In particular, we don't
-want them to worry about things like the execution environment and how
-their code is loaded into it.
-
-The Flow
-========
-
-We use prefect_ to express ETL pipelines. It provides us a few things:
-
-1. A Highlevel API for writing data transformations.
-2. A robust scheduling and orchestration system for executing pipelines.
-
-Every pangeo-forge recipe *must* have a ``prefect.Flow`` instance at the
-top-level of their pipeline module.
-
-
-.. code-block:: python
-
- class Pipeline(pangeo_forge.AbstractPipeline):
- @property
- def flow(self) -> Flow:
- with Flow(
- self.name,
- environment=self.environment,
- storage=self.storage
- ) as flow:
- # Your pipeline goes here.
- ...
-
- return flow
-
- pipeline = Pipeline()
- flow = pipeline.flow
-
-The ``storage`` keyword controls how your source code is loaded into
-pangeo-forge, and the ``environment`` keyword controls where it's run.
-"""
-import textwrap
-from abc import ABC, abstractmethod
-from pathlib import Path
-from typing import List
-
-from prefect import Flow
-from prefect.environments import DaskKubernetesEnvironment, Environment
-from prefect.environments.storage import Storage
-from prefect.environments.storage.github import GitHub
-
-HERE = Path(__file__).parent.absolute()
-
-
-def finalize_flow(cls):
- # Nothing to see here, move along.
- # OK, fine... this is a small hack to hide Prefect's orchestration stuff
- # (environments, storage) from the recipe writer. Prefect currently
- # requires that the module stored on GitHub have
- # 1. A `Flow` instance at the top-level of the module.
- # 2. A properly set `storage` and `environment`.
- # Which would require this at the bottom of every python module
- # >>> pipeline = MyPipeline()
- # >>> flow = pipeline.flow
- # >>> flow.storage = pipeline.storage
- # >>> flow.environment = pipeline.environment
- # Which is ugly. To avoid that, we just override how `.` works on
- # AbstractPipeline. If we're getting `.flow`, this little decorator
- # takes over and attaches the storage and environment.
- orig_getattribute = cls.__getattribute__
-
- def new_getattribute(self, name):
- result = orig_getattribute(self, name)
- if name == "flow":
- result.storage = self.storage
- result.environment = self.environment
- return result
-
- cls.__getattribute__ = new_getattribute
- return cls
-
-
-@finalize_flow
-class AbstractPipeline(ABC):
- name = "AbstractPipeline"
- path = "recipe/pipeline.py"
-
- @property
- @abstractmethod
- def repo(self):
- """The GitHub repository containing the pipeline definition."""
-
- @property
- @abstractmethod
- def sources(self) -> List[str]:
- """A list of source URLs containing the original data."""
- pass
-
- @property
- @abstractmethod
- def targets(self) -> List[str]:
- """A list of target URLs where the transformed data is written."""
- pass
-
- @abstractmethod
- def flow(self) -> Flow:
- """The """
- pass
-
- @property
- def environment(self) -> Environment:
- """
- The pipeline runtime environment.
-
- Returns
- -------
- prefect.environments.Environment
- An instance of a Prefect Environment. By default
- a :class:`prefect.environments.DaskKubernetesEnvironment`
- is used.
- """
- scheduler_spec_file = str(HERE / "job.yaml")
- worker_spec_file = str(HERE / "worker_pod.yaml")
-
- environment = DaskKubernetesEnvironment(
- min_workers=1,
- max_workers=30,
- scheduler_spec_file=scheduler_spec_file,
- worker_spec_file=worker_spec_file,
- metadata=dict(image="pangeoforge/default-image"),
- )
- return environment
-
- @property
- def storage(self) -> Storage:
- """
- The pipeline storage.
-
- Returns
- -------
- prefect.environments.storage.Storage
- By default a :class:`prefect.environments.storage.github.GitHub`
- environment is used with ``self.repo`` as the repository
- and ``self.path`` as the path.
- """
- return GitHub(self.repo, path=self.path)
-
- def _generate_run(self, source):
- name = type(self).__name__
- with open(source, encoding="utf-8") as f:
- source = f.read()
-
- return textwrap.dedent(
- """\
- # file: run.py
- # generated by pangeo-forge.
-
- {source}
-
- # ----------------------------------------------------------------
- pipe = {name}()
- flow = pipe.flow
- flow.storage = pipe.storage
- flow.environment = pipe.environment
- """
- ).format(source=source, name=name)

DELETE
pangeo_forge/pipelines/http_xarray_zarr.py
-from prefect import Flow
-
-from ..tasks.http import download
-from ..tasks.xarray import combine_and_write
-from ..tasks.zarr import consolidate_metadata
-from ..utils import chunked_iterable
-
-
-class HttpXarrayZarrMixin:
- @property
- def flow(self):
-
- if len(self.targets) == 1:
- target = self.targets[0]
- else:
- raise ValueError("Zarr target requires self.targets be a length one list")
-
- with Flow(self.name) as _flow:
-
- cached_sources = [download(k, self.cache_location) for k in self.sources]
-
- write_tasks = []
- for source_group in chunked_iterable(cached_sources, self.files_per_chunk):
- write_task = combine_and_write(
- source_group, target, self.append_dim, self.concat_dim,
- )
- write_tasks.append(write_task)
- cm = consolidate_metadata(target, write_tasks)
-
- # create dependencies in imperative mode
- for n in range(1, len(write_tasks)):
- write_tasks[n].set_upstream(write_tasks[n - 1], flow=_flow)
- cm.set_upstream(write_tasks[-1], flow=_flow)
-
- return _flow

DELETE
pangeo_forge/pipelines/job.yaml
-apiVersion: batch/v1
-kind: Job
-metadata:
- name: prefect-dask-job
- labels:
- app: prefect-dask-job
-spec:
- template:
- metadata:
- labels:
- app: prefect-dask-job
- spec:
- serviceAccountName: pangeo-forge
- serviceAccount: pangeo-forge
- containers:
- - name: flow
- imagePullPolicy: Always
- command: ["/bin/sh", "-c"]
- args:
- [
- 'python -c "import prefect; prefect.environments.execution.load_and_run_flow()"',
- ]
- resources:
- requests:
- cpu: "100m"
- limits:
- cpu: "100m"
- restartPolicy: Never

DELETE
pangeo_forge/pipelines/worker_pod.yaml
-kind: Pod
-metadata:
- labels:
- app: prefect-dask-worker
-spec:
- restartPolicy: Never
- serviceAccountName: pangeo-forge
- serviceAccount: pangeo-forge
- containers:
- - image: pangeoforge/noaa-oisst-avhrr
- imagePullPolicy: Always
- args: [dask-worker, --no-bokeh, --death-timeout, '60']
- name: dask-worker
- env:
- - name: PREFECT__CLOUD__GRAPHQL
- value: PREFECT__CLOUD__GRAPHQL
- - name: PREFECT__CLOUD__AUTH_TOKEN
- value: PREFECT__CLOUD__AUTH_TOKEN
- - name: PREFECT__CONTEXT__FLOW_RUN_ID
- value: PREFECT__CONTEXT__FLOW_RUN_ID
- - name: PREFECT__CLOUD__USE_LOCAL_SECRETS
- value: "false"
- - name: PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS
- value: "prefect.engine.cloud.CloudFlowRunner"
- - name: PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS
- value: "prefect.engine.cloud.CloudTaskRunner"
- - name: PREFECT__ENGINE__EXECUTOR__DEFAULT_CLASS
- value: "prefect.engine.executors.DaskExecutor"
- - name: PREFECT__LOGGING__LOG_TO_CLOUD
- value: "true"
- - name: PREFECT__LOGGING__LEVEL
- value: "DEBUG"
- - name: PREFECT__DEBUG
- value: "true"
- - name: PREFECT__LOGGING__EXTRA_LOGGERS
- value: PREFECT__LOGGING__EXTRA_LOGGERS
- resources:
- requests:
- cpu: "500m"
- limits:
- cpu: "500m"

DELETE
pangeo_forge/tasks/README.md
-# `pangeo_forge.tasks`
-
-A collection of Prefect Tasks for Pangeo-Forge.
-
-## Details
-
-- This is simply a collection of Prefect Tasks. Much like https://docs.prefect.io/api/latest/#task-library or https://github.com/PrefectHQ/prefect/tree/master/src/prefect/tasks.
-- This should be considered low-level public API.

DELETE
pangeo_forge/tasks/__init__.py
pangeo_forge/tasks/__init__.py
DELETE
pangeo_forge/tasks/http.py
-import os
-
-import fsspec
-from prefect import task
-
-
-@task
-def download(source_url, cache_location):
- """
- Download a remote file to a cache.
-
- Parameters
- ----------
- source_url : str
- Path or url to the source file.
- cache_location : str
- Path or url to the target location for the source file.
-
- Returns
- -------
- target_url : str
- Path or url in the form of `{cache_location}/hash({source_url})`.
- """
- target_url = os.path.join(cache_location, str(hash(source_url)))
-
- # there is probably a better way to do caching!
- 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())
- return target_url

DELETE
pangeo_forge/tasks/xarray.py
-from typing import List
-
-import fsspec
-import xarray as xr
-from prefect import task
-
-
-@task
-def combine_and_write(
- sources: List[str], target: str, append_dim: str, concat_dim: str
-) -> List[str]:
- """
- Write a batch of intermediate files to a combined zarr store.
-
- Parameters
- ----------
- sources : List[str]
- A list of URLs pointing to the intermediate files.
- target : str
- The URL for the target combined store.
- append_dim : str
- Name of the dimension of which datasets should be appended during write.
- concat_dim : str
- The dimension to concatenate along.
-
- Returns
- -------
- target : str
- The URL of the written combined Zarr store (same as target).
-
- Examples
- --------
- >>> import pangeo_forge.tasks.xarray
- >>> import fsspec
- >>> import xarray as xr
- >>> from prefect import Flow
-
- >>> # Load sample data into `sources`.
- >>> ds = xr.tutorial.open_dataset('rasm').load()
- >>> fs = fsspec.get_filesystem_class("memory")()
- >>> dsets = ds.isel(time=slice(18)), ds.isel(time=slice(18, None))
- >>> for i, dset in enumerate(dsets):
- ... as_bytes = dset.to_netcdf()
- ... with fs.open(f"cache/{i}.nc", "wb") as f:
- ... f.write(as_bytes)
-
- >>> sources = [f"memory://{dset}" for dset in fs.ls("cache")]
- >>> with Flow("my-flow") as flow:
- ... result = pangeo_forge.tasks.xarray.combine_and_write(
- ... sources, "memory://target.zarr", concat_dim="time"
- ... )
- >>> result
- <Task: combine_and_write>
-
- We can run that outside of a flow context with ``.run()``
- >>> pangeo_forge.tasks.xarray.combine_and_write.run(
- ... sources, "memory://target.zarr", concat_dim="time"
- ... )
- 'memory://target.zarr'
- """
- double_open_files = [fsspec.open(url).open() for url in sources]
- ds = xr.open_mfdataset(double_open_files, combine="nested", concat_dim=concat_dim)
- # by definition, this should be a contiguous chunk
- ds = ds.chunk({append_dim: len(sources)})
- mapper = fsspec.get_mapper(target)
-
- if not len(mapper):
- # The first write, .
- kwargs = dict(mode="w")
- else:
- kwargs = dict(mode="a", append_dim=append_dim)
- ds.to_zarr(mapper, **kwargs)
- return target

DELETE
pangeo_forge/tasks/zarr.py
-from typing import List, Optional
-
-import fsspec
-import zarr
-from prefect import task
-
-
-@task
-def consolidate_metadata(target, writes: Optional[List[str]] = None) -> None:
- """
- Consolidate the metadata the Zarr group at `target`.
-
- Parameters
- ----------
- target : str
- The URL for the (combined) Zarr group.
- writes : list of strings, optional
- The URLs the combined stores were written to. This is only a
- parameter to introduce a dependency in the pipeline execution graph.
- The actual value isn't used.
- """
- mapper = fsspec.get_mapper(target)
- zarr.consolidate_metadata(mapper)

prune old stuff