Skip to main content

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 Flux protocols serve slices of your data for you to process locally, openEO inverts this: you describe the analysis with the standard openEO Python client, Flux runs it next to your Arraylake data, and you download only the result.

References​

These are links to the openEO standard and websites:

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.

Activating openEO for Arraylake Datasets​

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

al compute enable {org} openeo

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 Flux 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.

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.

ProcessNotes
load_collectionOpen a collection as a datacube. Supports bands, temporal_extent (inclusive [start, end]), and spatial_extent (bounding box).
aggregate_spatialSample the datacube at GeoJSON Point geometries, returning a vector datacube.
meanThe only supported aggregate_spatial reducer.
save_resultSerialize the result to parquet, csv, geojson, or netcdf.

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-09-30"],
)

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.

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()

results = job.get_results()
results.download_files("output/")

Or load the result directly into pandas:

from io import BytesIO

table = pd.read_parquet(BytesIO(results.get_assets()[0].load_bytes()))

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

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.