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}")
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.
| Process | Notes |
|---|---|
load_collection | Open a collection as a datacube. Supports bands, temporal_extent (inclusive [start, end]), and spatial_extent (bounding box). |
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. |
mean | The only supported aggregate_spatial reducer. |
save_result | Serialize 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.
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()
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()))
A large job may be fanned out across many workers. When that happens, the results are split and results.get_assets() returns a list with one file per part rather than a single file. Iterate over the assets (or use results.download_files(), which fetches all of them) 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.
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://{bucket_nickname}/{key} path, where key is relative to the bucket's configured prefix:
points = connection.datacube_from_process(
"load_uploaded_files",
paths=["arraylake://{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.