openEO: Server-Side Processing

openEO is an open API standard for describing geospatial analyses as process graphs and executing them on the backend where the data lives. Where the other Compute protocols serve slices of your data for you to process locally, openEO inverts this: you describe the analysis with the standard openEO Python client, Compute runs it next to your Arraylake data, and you download only the result.

References

These are links to the openEO standard and websites:

• openEO Main Page
• openEO API Specification
• openEO Python Client

Arraylake implements a subset of the openEO API (v1.2.0): collection discovery, synchronous processing, and batch jobs, with a process set currently focused on sampling points from gridded datasets, band math, and reducing dimensions.

Activating openEO for Arraylake Datasets

openEO can be activated using the Arraylake command line interface (CLI)

al compute enable {org} openeo --sku-code {sku}

The service is enabled on an organization-wide basis; every Repo the caller has access to is served as a collection.

Connecting

The easiest way to connect is with the arraylake Python client, which resolves the service URL and authenticates the connection for you:

from arraylake import Client

client = Client()
connection = client.get_openeo_connection("{org}")

note

get_openeo_connection requires arraylake>=1.2.0 and the optional openeo dependency, installed with pip install 'arraylake[openeo]'.

Any stock openEO client can also connect directly using the service URL:

import openeo

connection = openeo.connect("https://compute.earthmover.io/v1/services/openeo/{org}")
connection.authenticate_bearer_token("{api_token}")

Unlike the other Compute protocols, openEO clients authenticate with a bearer token — use an Arraylake API token — rather than HTTP Basic auth. Public services require no authentication.

Collections

Datasets are addressed as openEO collections using an Arraylake path as the collection id:

{org}/{repo}
{org}/{repo}/{branch|commit|tag}
{org}/{repo}/{branch|commit|tag}/{path/to/group}

The ref defaults to main and the group defaults to the root group.

connection.list_collections()
connection.describe_collection("{org}/{repo}")

describe_collection returns a STAC collection describing the datacube's dimensions (cube:dimensions) and spatial and temporal extents. As with the other protocols, this works best when datasets are compliant with CF Conventions, particularly around coordinates and CRS metadata. Coordinates named latitude/lat and longitude/lon with no CF attributes are still recognized as WGS84, as long as their values are valid latitudes and longitudes.

Supported Processes

The service advertises its process set at GET /processes. Submitted graphs are validated against this set, so graphs using unsupported processes are rejected before anything runs.

Process	Notes
load_collection	Open a collection as a datacube. Supports bands, temporal_extent, and spatial_extent (bounding box), which apply the three filters below.
filter_bands	Keep only the named bands.
filter_temporal	Keep the left-closed interval [start, end) of a temporal dimension. Either bound may be null for an open interval.
filter_bbox	Keep a bounding box, given in WGS84 unless its crs says otherwise.
filter_labels	Keep the labels of a dimension that a condition accepts, e.g. one forecast lead time or a few ensemble members. Compare the label value with constants: use eq or array_contains (joined with or) for exact labels, or gt, gte, lt, lte or between (joined with and) for a range. Lead times are written as ISO 8601 durations such as "PT6H".
load_uploaded_files	Load point geometries from files previously uploaded to one of your organization's buckets, referenced by arraylake:// paths. See Large Geometry Sets.
aggregate_spatial	Sample the datacube at Point geometries — inline GeoJSON or a vector cube from load_uploaded_files — returning a vector datacube.
apply	Apply elementwise math (see the callback math below) to every value of every band.
apply_dimension	Over bands: append computed bands, or replace the bands with them. To append, the callback returns its data extended with array_append. The openEO client's append_index and compute_indices helpers also work. See Derived Bands.
rename_labels	Rename bands: source[i] becomes target[i].
merge_cubes	Combine the bands of two datacubes, such as two collections on the same grid. The cubes must share their grid already: load them with the same spatial_extent and temporal_extent, and apply the same filters to both, because merging doesn't resample. The cubes can't have bands in common, and they must be merged before aggregate_spatial.
add_dimension	add_dimension(name="bands", label=..., type="bands") names the band that a band reduction (reduce_dimension over bands, or ndvi without target_band) left. Until then, processes that work on bands (such as filter_bands) fail with DimensionNotAvailable.
reduce_dimension	Reduce a dimension of the datacube. bands reduces with band math (array_element and the callback math below) into one band named value (name it with add_dimension); any other dimension (an ensemble, a lead time, time) reduces with mean, sum, sd, or variance, each band keeping its name. See Ensemble Statistics.
ndvi	Compute (nir - red) / (nir + red) from the named bands: one band named ndvi, or a new band named by target_band beside the others.
mean, median, min, max, sum, first, last	Reducers. aggregate_spatial accepts any of them; over a Point each is the sampled cell's value. mean and sum also reduce in reduce_dimension. count is not supported.
sd, variance	The sample standard deviation and variance, as reduce_dimension reducers.
array_element, array_append, array_create, array_modify, linear_scale_range, add, subtract, multiply, divide, power, normalized_difference, absolute, sqrt, cos	Callback math for apply, apply_dimension and reduce_dimension. array_element picks a band by index or label, array_append adds a computed band, and linear_scale_range clips and rescales values. Integer bands are promoted to float64. A computed band gets a units attribute when it follows from its inputs' units; adding, subtracting, multiplying or dividing by a constant, which may be a unit conversion, leaves it unset.
eq, array_contains, between, gt, gte, lt, lte, and, or	Conditions for filter_labels.
save_result	Serialize the result to parquet, csv, geojson, or netcdf. A datacube can also be saved without aggregating it, subject to the same result size limit; netcdf preserves its dimensions. Batch jobs can instead commit the result to a Repo with format icechunk (see Saving Results to a Repo).

Temporal intervals follow the openEO standard: the end is excluded, so ["2024-09-01", "2024-10-01"] selects all of September.

Example: Sampling Points from a Datacube

The canonical openEO workflow on Arraylake extracts values at many point locations from a large gridded dataset, returning a small table instead of the underlying chunks.

Build the process graph by chaining datacube operations:

import pandas as pd

points = pd.read_csv("points.csv")  # columns: id, longitude, latitude

geometries = {
    "type": "FeatureCollection",
    "features": [
        {
            "type": "Feature",
            "properties": {"id": row.id},
            "geometry": {
                "type": "Point",
                "coordinates": [row.longitude, row.latitude],
            },
        }
        for row in points.itertuples(index=False)
    ],
}

cube = connection.load_collection(
    "{org}/{repo}",
    bands=["t2m"],
    temporal_extent=["2024-09-01", "2024-10-01"],
)

result = cube.aggregate_spatial(geometries=geometries, reducer="mean").save_result(format="parquet")

Nothing has executed yet — result is a process graph that can be run synchronously or as a batch job.

note

Inline GeoJSON works up to a few thousand points; beyond that the request exceeds the graph size cap and is rejected. For larger point sets, see Large Geometry Sets.

Synchronous Execution

Small requests can be executed synchronously, streaming the result back within the request:

result.download("samples.parquet")

Requests that take too long to run synchronously are rejected with an error directing you to submit a batch job instead.

Batch Jobs

Larger requests should run as batch jobs, which execute durably in the background and write their results to your organization's bucket:

job = result.create_job(title="t2m point sampling")
job.start_and_wait()

Result assets are presigned object-storage URLs, so fetch them with a plain HTTP GET. Don't use results.download_files() or asset.download(): the openEO client sends its bearer token to the presigned URL, and the storage service rejects the request with a 403.

from pathlib import Path

import requests

out = Path("output")
out.mkdir(exist_ok=True)
for name, asset in job.get_results().get_metadata()["assets"].items():
    response = requests.get(asset["href"], timeout=600)  # no Authorization header
    response.raise_for_status()
    (out / name).write_bytes(response.content)

table = pd.concat(pd.read_parquet(path) for path in sorted(out.glob("*.parquet")))

note

A large job may be fanned out across many workers. When that happens, the results are split into one asset per part rather than a single file. Fetch every asset, as above, instead of assuming a single result file.

Jobs can be listed with connection.list_jobs(), and a finished job's results remain downloadable until the job is deleted.

Example: Ensemble Statistics

reduce_dimension computes statistics along any dimension of a datacube, such as the members of an ensemble forecast. This example derives 10 m wind speed from its components, then takes its ensemble spread:

cube = connection.load_collection(
    "{org}/{repo}",
    bands=["wind_u_10m", "wind_v_10m"],
    temporal_extent=["2026-10-01", "2026-10-02"],
)
speed = cube.reduce_dimension(
    dimension="bands",
    reducer=lambda data: (
        data.array_element(label="wind_u_10m") ** 2
        + data.array_element(label="wind_v_10m") ** 2
    ).sqrt(),
)
spread = speed.reduce_dimension(
    dimension="ensemble_member",
    reducer=lambda data: data.sd(ignore_nodata=False),
)
result = spread.aggregate_spatial(geometries=geometries, reducer="mean").save_result(format="parquet")

A few rules apply when reducing a dimension other than bands:

• The reducer is one of mean, sum, sd, or variance, applied to data directly, and must pass ignore_nodata=False: a missing value anywhere along the dimension makes the result missing.
• Every band must have the dimension, so select the bands to reduce with bands or filter_bands first.
• A reduction reads every value it folds: an ensemble mean over 51 members reads 51 times its output. The result size limit counts only the result, but the extra reading takes time, so keep the other dimensions (time, area, points) bounded.
• sd and variance are computed in one pass, so they lose precision on float32 data whose values are large relative to their spread (temperatures in kelvin, for example).

Example: Derived Bands

apply_dimension over bands adds a computed band beside the bands it was computed from. This example appends 10 m wind speed to its components, keeping the 6-hour lead time and three ensemble members:

cube = connection.load_collection(
    "{org}/{repo}",
    bands=["wind_u_10m", "wind_v_10m"],
    temporal_extent=["2026-10-01", "2026-10-02"],
)
cube = cube.filter_labels(condition=lambda lead: lead == "PT6H", dimension="lead_time")
cube = cube.filter_labels(
    condition=lambda member: openeo.processes.array_contains([0, 1, 2], member),
    dimension="ensemble_member",
)
cube = cube.apply_dimension(
    dimension="bands",
    process=lambda data: data.array_append(
        (
            data.array_element(label="wind_u_10m") ** 2
            + data.array_element(label="wind_v_10m") ** 2
        ).sqrt(),
        label="wind_speed_10m",
    ),
)

The result has three bands: wind_u_10m, wind_v_10m, and wind_speed_10m. The new band carries the units of its inputs (m s-1) when the dataset records them. To keep only the new band, follow with filter_bands(["wind_speed_10m"]).

Saving Results to a Repo

A batch job can commit its result to a new group of an Arraylake Repo instead of writing a file, with save_result format icechunk. The result is written straight from the datacube, chunk by chunk, so it isn't subject to the result size limit, and it's committed once, whole: if anything fails, nothing is committed.

result = cube.save_result(
    format="icechunk",
    options={
        "repo": "{org}/{results-repo}",  # required
        "branch": "main",                # default: main
        "group": "wind/speed-10m",       # the new group to write; default: the root group
        "message": "10 m wind speed",    # the commit message
    },
)
job = result.create_job(title="10 m wind speed")
job.start_and_wait()

• You need write access to the target Repo, as well as read access to the Repos the graph reads. On a public deployment, authenticate with your own token to save.
• The group must not exist yet. Saving to the root group requires an empty Repo. A result never overwrites existing data.
• Saving to a Repo runs only as a batch job; synchronous execution refuses it.
• The result is stored the way its source was: data it kept keeps its source's chunking and compression, and computed bands are chunked along the same dimensions and compressed with Zstandard.
• The commit message names the job and run that made the commit. A job re-delivered after it committed reuses that commit rather than writing a second one.

The finished job's results have one asset, the collection it committed, pinned to the commit's snapshot. Its arraylake:collection_id loads exactly what the job wrote:

asset = job.get_results().get_metadata()["assets"]["result"]
saved = connection.load_collection(asset["arraylake:collection_id"])

The job itself reports where it committed under arraylake:result_commit: the Repo, branch, group and snapshot id.

Large Geometry Sets

Process graphs travel inside a size-capped request, which limits inline GeoJSON geometries to a few thousand points. For larger point sets, upload the geometry file to one of your organization's buckets and reference it with the load_uploaded_files process instead of inlining it.

Upload the file with the Arraylake client's obstore helper, which scopes writes to the bucket and its configured prefix:

import obstore

store = await client.aclient.get_obstore_for_bucket(
    org="{org}", nickname="{bucket_nickname}", access="write"
)
with open("points.csv", "rb") as f:
    await obstore.put_async(store, "openeo-uploads/points.csv", f.read())

Then reference it as an arraylake://{org}/bucket/{bucket_nickname}/{key} path, where org is the organization that owns the compute service you submit to and key is relative to the bucket's configured prefix:

points = connection.datacube_from_process(
    "load_uploaded_files",
    paths=["arraylake://{org}/bucket/{bucket_nickname}/openeo-uploads/points.csv"],
    format="csv",
)

result = cube.aggregate_spatial(geometries=points, reducer="mean").save_result(format="parquet")
job = result.create_job(title="large point sampling")

Supported formats are geojson, csv, and parquet. For tabular formats, coordinate columns named lon/longitude/x and lat/latitude/y are detected case-insensitively; other names can be given explicitly with options={"x_column": "...", "y_column": "..."}. Multiple paths are concatenated in order.

Using load_uploaded_files requires the org-wide CAN_READ_REPOS permission, since the referenced files are read with the organization's bucket credentials.

Result Artifacts

Every execution writes its result as an object, an artifact, in one of your organization's bucket configurations, under a reserved prefix:

{bucket_prefix}/__earthmover_artifacts__/openeo_artifacts/{execution_id}/

By default, artifacts are written to the bucket where the Repo referenced by the graph's load_collection lives. Batch jobs can target a different bucket in your organization by passing its bucket config nickname in the bucket_nickname field (a non-standard extension) when creating the job:

job = result.create_job(
    title="t2m point sampling",
    additional={"bucket_nickname": "scratch"},
)

Synchronous executions stage their result the same way before streaming it back, always using the default bucket.

The asset hrefs in a finished job's results are presigned object-storage URLs, so any HTTP client can download them directly without a token. Deleting a job removes it from the API but does not remove its artifacts from the bucket; they can be managed like any other objects there.