Skip to main content

fused.h3

delete_index​

delete_index(
    path: str,
    *,
    prefix: str | list[str] | None = None,
    base_url: str | None = None
) -> dict[str, Any]

Delete the index registered for path, or with prefix only the part of it under those folders. The data files are untouched.

Parameters:

  • path (str) – The location an earlier fused.h3.index call registered.
  • prefix (str | list[str] | None) – Optional. A folder under path (relative to it, such as "month=1940-01", or spelled in full), or a list of them. Only what is indexed under those folders is deleted; the location keeps serving the rest. Folders are cleared one at a time, so a call that is interrupted leaves each folder either fully removed or intact, and calling again finishes the rest. A later fused.h3.index run that still lists those files indexes them again. Without prefix the whole index is deleted, location included.
  • base_url (str | None) – Base URL for the API. If None, uses the current environment.

Returns:

  • dict[str, Any] – A dictionary with the resolved path, and for a scoped delete the full prefix paths, the count of index rows deleted and the number of folders cleared.

get_run​

get_run(run_id: str, *, base_url: str | None = None) -> TaskRun | Run

The run with id run_id, in a new session or process.

A Run for the id of a run of several jobs (grp:...), as fused.h3.partition returns, otherwise the TaskRun of that single run.

Raises:

  • HTTPError – There is no such run in your environment (404).

index​

index(
    path: str,
    hex_column: str = "hex",
    time_column: str | None = None,
    *,
    partition_columns: str | list[str] | None = None,
    subset: dict[str, Any] | None = None,
    redo: bool = False,
    engine: InstanceType | Literal["realtime"] = "realtime",
    visibility: str | None = None,
    options: dict[str, Any] | None = None,
    base_url: str | None = None,
    **deprecated: Any
) -> TaskRun

Register a hive-partitioned parquet tree as a queryable H3 location.

For a tree written by fused.h3.partition, fused.h3.index(output) needs no other arguments. It starts a server-side run that walks path and builds the row-group index, and returns as soon as the run has started; follow it with run.tail() or index_status. The index belongs to your team; see visibility and fused.h3.set_visibility for sharing it.

Running it again is incremental: folders unchanged since the last run are skipped, and only new, changed and removed files are indexed or dropped. A run that was stopped or partly failed resumes where it left off.

Parameters:

  • path (str) – Path to the parquet tree (e.g. "s3://my-bucket/my-dataset/"). Every .parquet file under it, at any depth, must carry an integer H3 cell column named hex_column at a single resolution.
  • hex_column (str) – Name of the H3 cell column present in every file. Defaults to "hex", what fused.h3.partition writes.
  • time_column (str | None) – Optional. Names either an in-file timestamp/date column, or a folder level written key=value whose value is a YYYY, YYYY-MM, or YYYY-MM-DD period covering every row under that folder, or an ISO timestamp (an instant). If None, the only date partition column filters by time, and a location without one is timeless.
  • partition_columns (str | list[str] | None) – The key=value folder levels (e.g. scenario=ssp245/date=2050-01/) whose values query can filter by, as a column name or a list of names in folder order. Defaults to the partition_by columns in the tree's _manifest.json, else none. Each leaf folder must have exactly one folder per column, in that order.
  • subset (dict[str, Any] | None) – Limit the run to some folders, as {column: value or list of values}. Unlike query(partition=...), the keys must be the leading partition columns of the tree (a column cannot be skipped), in any order. Values are strings, numbers or booleans. Only those folders are listed and indexed, so adding one month to a large archive costs that month.
  • redo (bool) – Read files again. With subset, every file in those folders is read again and replaces what was indexed from it; nothing else is touched. Without it, the whole tree is indexed from scratch, which is also how to change the location's columns.
  • engine (InstanceType | Literal['realtime']) – "realtime" (the default) starts at once but is limited to a few minutes. A batch instance type or alias ("small", "medium", "large", ...) has no time limit; use it for a large tree.
  • visibility (str | None) – "team" or "public" (Fused admins only); None keeps the location's current visibility ("team" for a new one). A public location's files must be publicly readable on S3.
  • options (dict[str, Any] | None) – Tuning, as a dict. "workers": how many files are read at once (default 32; a larger instance can take more).
  • base_url (str | None) – Base URL for the API. If None, uses the current environment.

Returns:

  • TaskRun – The TaskRun of the indexing run. run.tail() prints its log lines until it finishes, run.status() and run.done() poll it, and index_status(path) reports the index once it has succeeded. If only some files fail, everything else is indexed and queryable; the run's error names the failed paths, and running again retries them.

Example:

run = fused.h3.index("s3://my-bucket/era5_h3/")
run.tail()

Add a month, then redo it:

fused.h3.index("s3://my-bucket/era5_h3/", subset={"month": "2026-11"})
fused.h3.index(
    "s3://my-bucket/era5_h3/", subset={"month": "2026-11"}, redo=True
)

Deprecated: prefix= (use subset=), overwrite=True and force=True (use redo=True) and workers= (use options) still work and warn.

index_status​

index_status(
    path: str, *, base_url: str | None = None
) -> dict[str, Any] | None

Describe the index registered for path, or None when path was never indexed and no run is queued or running for it.

Parameters:

  • path (str) – The location an earlier fused.h3.index call registered.
  • base_url (str | None) – Base URL for the API. If None, uses the current environment.

Returns:

  • dict[str, Any] | None – None when there is neither an index nor a run for path. Otherwise a dictionary with path, status ("queued" for a never-indexed path whose job has not started, "indexing", "ready", or "failed", which for a path that never finished a run may also mean the job died before starting, with its error in failures), indexing (True while a run is queued or running), job_id (the batch job of that run, or None), files, resolution, hex_column, time_column, partition_columns (per column: name, type, distinct, min, max), time_min, time_max, indexed_as_of (files modified after this may not be indexed yet), last_run_mode ("incremental" or "overwrite"), last_run_finished_at, and failures (a list of {path, error} for files the last run could not index).

indexed_files​

indexed_files(
    path: str, *, prefix: str | None = None, base_url: str | None = None
) -> Iterator[str]

Iterate the files indexed under path, in path order.

Parameters:

  • path (str) – The location an earlier fused.h3.index call registered.
  • prefix (str | None) – Only yield files under this folder or path prefix.
  • base_url (str | None) – Base URL for the API. If None, uses the current environment.

Yields:

  • str – The full path of each indexed file.

partition​

partition(
    data: Any,
    output: str | Path,
    *,
    hex_column: str | None = None,
    point_column: str | tuple[str, str] | None = None,
    h3_res: int = 7,
    file_res: int = 1,
    fill: str | None = None,
    partition_by: (
        str | list[str] | dict[str, Any] | list[dict[str, Any]] | None
    ) = None,
    sort_by: str | list[str] | None = None,
    redo: bool = False,
    chunk_res_offset: int | None = None,
    read_engine: InstanceType = "small",
    engine: InstanceType | Literal["realtime"] = "small",
    options: dict[str, Any] | None = None,
    base_url: str | None = None,
    **deprecated: Any
) -> Run

Write data as hex-sorted H3 parquet files with a _manifest.json.

The work runs on the Fused server and this returns a Run as soon as it has started: run.tail() follows it. A udf or era5 source is read first, then partitioned; a parquet source is partitioned directly.

Example:

An ERA5 year, one job per month::

    run = fused.h3.partition(
        fused.h3.sources.era5(outputs=["t2m_mean"]),
        "fd://me/era5_2024/",
        partition_by={"month": [f"2024-{m:02d}" for m in range(1, 13)]},
        point_column=("lat", "lng"), fill="nearest",
        read_engine="large",
    )
    run.tail()
Details

Recipes Add a month: run it again with the new value in partition_by. The new folders are written and the output is merged; jobs already done are skipped.

Redo a month: pass redo=True with just that month. It is read again from the source and rewritten.

Retry failures: run.retry() runs the failed jobs again.

Change settings (h3_res, file_res, fill, sort_by, partition_by columns): write to a new output, or delete the existing output folder. A run whose settings differ from the existing output raises and says so.

Parameters:

  • data (Any) – A fused.h3.sources source (udf, era5, parquet) or a path on s3://, gs://, fd:// or /mount/<folder>, read like sources.parquet(path). To partition data you have in Python, return it from a UDF. Nothing is read from this machine.
  • output (str | Path) – Output folder on fd://, s3://, gs:// or the team's /mount/<folder>. Files are <output>/<col>=<value>/.../<H3 parent cell>.parquet. Local paths and az:// are not supported. Leftover files without a _manifest.json raise.
  • hex_column (str | None) – Name of an existing H3 cell column (uint64). Give exactly one of hex_column and point_column.
  • point_column (str | tuple[str, str] | None) – ("lat_column", "lng_column") in degrees, or the name of one WKB geometry column (a non-point gets its centroid's cell). One row in, one row out; nothing is polyfilled.
  • h3_res (int) – Resolution of each row's cell, the hex column queries select by.
  • file_res (int) – Resolution of the parent cell that groups rows into one file (default 1; raise it for more, smaller files).
  • fill (str | None) – "nearest" fills every cell at h3_res from the nearest point of a regular lat/lng raster (as sources.era5 yields) instead of hexing each row. Needs point_column=("lat", "lng").
  • partition_by (str | list[str] | dict[str, Any] | list[dict[str, Any]] | None) – What becomes col=value folders, and how the run is split into jobs. A column name or a list of names: folders from the data, one job. A dict {column: value or list of values}: one job per combination (the product, in key order), each bound on data (any iterable, a range too, is a list of values; a date is sent as ISO text) like sources.era5(month=...), a UDF's parameters or a sources.parquet hive folder. A list of dicts is an explicit job list, all with the same keys. The dict keys are the folder columns. A job is named by its folder, such as scenario=ssp245/month=2024-01.
  • sort_by (str | list[str] | None) – Extra column(s) to sort by after hex.
  • redo (bool) – Redo what this call covers. With partition_by jobs (a dict or list of dicts), the jobs are read again from the source and rewritten even if done; the default skips a job that is already done. A single job with partition_by columns is read again and merged into the output: the folders it writes are replaced, all other folders stay, and a folder the source no longer produces stays too (it is never wiped; to start over, write to a new output path or delete the output folder). It refuses an output written one job per value. A single job without partition_by replaces the output, since it covers all of it. Without redo, a single job with columns adds to the output in the same way, and one without columns raises if the output exists. Do not run a single job (with or without columns, or with redo=True) into an output while per-job runs (a dict partition_by) are writing to it.
  • chunk_res_offset (int | None) – Align the row groups to H3 parent cells: every row group holds whole cells of resolution file_res + chunk_res_offset (the chunk resolution, recorded in the manifest and in each file's footer as fused:chunk_res), and row_group_size is ignored. Each file gets about 7**chunk_res_offset row groups (an offset of 3 is about 343). Needs 0 <= chunk_res_offset and file_res + chunk_res_offset <= h3_res. It trades cost for reach: coarse chunks make a small query decode big row groups, fine chunks make many row groups (a bigger footer and more requests). None (the default) writes row groups of row_group_size rows and records no chunk resolution.
  • read_engine (InstanceType) – Batch instance type of the read jobs of a udf or era5 source (default "small"). "realtime" raises.
  • engine (InstanceType | Literal['realtime']) – "small" (default) runs the partition job as a batch job with no time limit; larger batch types give it more memory. "realtime" runs on the realtime tier for a few minutes at most, for small parquet inputs only.
  • options (dict[str, Any] | None) – Tuning, as a dict. row_group_size (65536; not with chunk_res_offset), compression ("zstd"), compression_level (5), key_columns (columns to dictionary-encode), memory_budget and workers (both sized from the instance by default). An unknown key raises.
  • base_url (str | None) – Fused API base URL; defaults to the configured one.

Returns:

  • Run – A Run. run.tail() prints each job's logs as [job] ...; run.status(), run.jobs(), run.errors(), run.failed(), run.succeeded() and run.skipped() say how it went; fused.h3.get_run(run.id) brings it back in a new session.
Details

Deprecated parts= (use partition_by), only= (pass only those values), force= and refresh= (use redo=True), append= (a rerun merges anyway), overwrite= (use a new output path) and the one-by-one row_group_size, compression, compression_level, key_columns, memory_budget and workers (use options) still work for one release and warn.

query​

query(
    path: str,
    cell: str | None = None,
    *,
    lat: float | None = None,
    lng: float | None = None,
    bbox: tuple[float, float, float, float] | None = None,
    hex_range: tuple[str, str] | None = None,
    time_min: str | None = None,
    time_max: str | None = None,
    partition: dict[str, Any] | None = None,
    partition_range: dict[str, Any] | None = None,
    columns: list[str] | None = None,
    max_row_groups: int | None = None,
    base_url: str | None = None,
    **kwargs: Any
) -> pd.DataFrame | Any

Read the rows of a previously indexed H3 location: your team's own index of path, else the public one.

Only available inside UDFs running on Fused (a realtime or batch instance). The rows are read in the calling process, so the whole result is held in its memory. To keep memory down, narrow the selector or time window (and fan out across narrower queries with fused.submit if more than one is needed).

At most one spatial selector may be given: cell (an H3 cell id, as a hex string -- matches every row filed under a descendant of cell, or under cell itself if it is already at the location's resolution), lat+lng (a point), bbox (a (min_lng, min_lat, max_lng, max_lat) tuple), or hex_range (an inclusive raw H3 cell-id range, as a (hex_min, hex_max) tuple of hex strings). Omitting all of them returns every row at the location.

Only space, time and partition values are filtered. Filter other columns (a vessel type, a LiDAR class) on the returned DataFrame. A result too large to hold in memory fails the job; split a larger area or time window into smaller queries.

Parameters:

  • path (str) – The location an earlier fused.h3.index call registered.
  • cell (str | None) – An H3 cell id (as a hex string, e.g. "8928308280fffff").
  • lat (float | None) – Latitude of a point selector. Requires lng.
  • lng (float | None) – Longitude of a point selector. Requires lat.
  • bbox (tuple[float, float, float, float] | None) – (min_lng, min_lat, max_lng, max_lat) bounding box selector. A row is kept when the centre of its cell lies inside the box, so the edge is accurate to one cell at the location's resolution. For an exact edge, pad the box by one cell and filter the point columns.
  • hex_range (tuple[str, str] | None) – (hex_min, hex_max) inclusive raw H3 cell-id range.
  • time_min (str | None) – Inclusive lower bound (ISO datetime). Only valid against a location with a time column (a time_column, or a single date partition column).
  • time_max (str | None) – Inclusive upper bound (ISO datetime). Only valid against a location with a time column.
  • partition (dict[str, Any] | None) – Keep only the folders whose partition values match, as {column: value}. A list or tuple of values matches any of them: {"scenario": ["ssp245", "ssp585"], "date": "2050-01"}. A folder must match every column named. A date column matches by overlap: "2050-01" matches a folder holding January 2050.
  • partition_range (dict[str, Any] | None) – Keep only the folders whose partition value lies in an inclusive (min, max) range, as {column: (min, max)}; None leaves an end open: {"date": ("2050-01", None)}. Works for string, number and date columns. A range on the location's time column also narrows the time window.
  • columns (list[str] | None) – Which columns to read. Defaults to all columns.
  • max_row_groups (int | None) – How much of the dataset one page of the result may read. It sets the page size, not a limit on the result: the query follows the pages until every row is read. If None (the default), a default page size applies.
  • base_url (str | None) – Base URL for the API. If None, uses the current environment.

Returns:

  • DataFrame | Any – A pandas DataFrame with the matching rows.

Raises:

  • RuntimeError – If called outside a UDF running on a Fused realtime or batch instance.
  • HTTPError – If the API request fails (404 if path has not been indexed, 409 if it is not yet "ready", if a data file changed after it was indexed (re-run fused.h3.index), or if it was re-indexed while the pages were being read (run the query again), 400 for an invalid or conflicting selector, against time bounds on a timeless location, or for a partition or partition_range naming a column that is not a partition column or holding a value that does not fit it).

Example:

df = fused.h3.query(
    path="s3://my-bucket/era5_h3/",
    cell="8928308280fffff",
    time_min="2024-01-01T00:00:00",
    time_max="2024-01-31T23:59:59",
)

df = fused.h3.query(
    path="s3://my-bucket/climate_h3/",
    bbox=(-122.5, 37.7, -122.4, 37.8),
    partition={"scenario": "ssp245"},
    partition_range={"date": ("2050-01", "2050-06")},
)

run_ingest_raster_to_h3​

run_ingest_raster_to_h3(
    input_path: str | list[str],
    output_path: str,
    metrics: str | list[str] = "cnt",
    res: int | None = None,
    k_ring: int = 1,
    res_offset: int = 1,
    chunk_res: int | None = None,
    file_res: int | None = None,
    overview_res: list[int] | None = None,
    overview_chunk_res: int | list[int] | None = None,
    max_rows_per_chunk: int = 0,
    include_source_url: bool = True,
    target_chunk_size: int | None = None,
    nodata: int | float | None = None,
    debug_mode: bool = False,
    remove_tmp_files: bool = True,
    tmp_path: str | None = None,
    overwrite: bool = False,
    steps: list[str] | None = None,
    extract_engine: str | None = None,
    extract_max_workers: int = 256,
    partition_engine: str | None = None,
    partition_max_workers: int = 256,
    overview_engine: str | None = None,
    overview_max_workers: int = 256,
    **kwargs: int
)

Run the raster to H3 ingestion process.

This process involves multiple steps:

  • extract pixels values and assign to H3 cells in chunks (extract step)
  • combine the chunks per partition (file) and prepare metadata (partition step)
  • create the metadata _sample file and overviews files

Parameters:

  • input_path (str, list) – Path(s) to the input raster data. When this is a single path, the file is chunked up for processing based on target_chunk_size. When this is a list of paths, each file is processed as one chunk.
  • output_path (str) – Path for the resulting Parquet dataset.
  • metrics (str or list of str) – The metrics to compute per H3 cell. Supported metrics are either "cnt" or a list containing any of "avg", "min", "max", "stddev", "mode" (i.e. most common value), and "sum".
  • res (int) – The resolution of the H3 cells in the output data. The pixel values are assigned to H3 cells at resolution res + res_offset and then aggregated to res. By default, this is inferred based on the resolution of the input data ensuring the H3 cell size is close to the pixel size (e.g. for a raster with pixel size of 30x30m, a resolution of 11 is inferred).
  • k_ring (int) – The k-ring distance at resolution res + res_offset to which the pixel value is assigned (in addition to the center cell). Defaults to 1.
  • res_offset (int) – Offset to child resolution (relative to res) at which to assign the raw pixel values to H3 cells.
  • file_res (int) – The H3 resolution to chunk the resulting files of the Parquet dataset. By default will be inferred based on the target resolution res. You can specify file_res=-1 to have a single output file.
  • chunk_res (int) – The H3 resolution to chunk the row groups within each file of the Parquet dataset (ignored when max_rows_per_chunk is specified). By default will be inferred based on the target resolution res.
  • overview_res (list of int) – The H3 resolutions for which to create overview files. By default, overviews are created for resolutions 3 to 7 (or capped at a lower value if the res of the output dataset is lower).
  • overview_chunk_res (int or list of int) – The H3 resolution(s) to chunk the row groups within each overview file of the Parquet dataset. By default, each overview file is chunked at the overview resolution minus 5 (clamped between 0 and the res of the output dataset).
  • max_rows_per_chunk (int) – The maximum number of rows per chunk in the resulting data and overview files. If 0 (the default), chunk_res and overview_chunk_res are used to determine the chunking.
  • include_source_url (bool) – If True, include a "source_url" column in the output dataset that contains a list of source URLs that contributed data to each H3 cell. Defaults to True, set to False to omit this column.
  • target_chunk_size (int) – The approximate number of pixel values to process per chunk in the first "extract" step. Defaults to 10,000,000 for ingesting a single file or a few files. If ingesting more than 20 files, each file is processed as a single chunk by default, but you can override this by specifying a specific target_chunk_size value, or by specifying target_chunk_size=0 to always process each file as a single chunk.
  • nodata (int or float) – By default, the "nodata" value indicated in the raster metadata will be used to ignore pixels with this value when ingesting the data. If this information is missing or wrong in the raster metadata, you can override the value by specifying this keyword.
  • debug_mode (bool) – If True, run only the first two chunks for debugging purposes. Defaults to False.
  • remove_tmp_files (bool) – If True, remove the temporary files after ingestion is complete. Defaults to True.
  • tmp_path (str) – Optional path to use for the temporary files. If specified, the extract step is skipped and it is assumed that the temporary files are already present at this path.
  • overwrite (bool) – If True, overwrite the output path if it already exists, by first removing the existing content before writing the new files. Defaults to False, in which case an error is raised if the output_path is not empty.
  • steps (list of str) – The processing steps to run. Can include "extract", "partition", "metadata", and "overview". By default, all steps are run.
  • extract_engine (str) – The engine to use for the extract step. By default, tries first with "realtime" and then falls back to "large".
  • extract_max_workers (int) – The maximum number of workers to use for the extract step. Defaults to 256.
  • partition_engine (str) – The engine to use for the partition step. By default, tries first with "realtime" and then falls back to "large".
  • partition_max_workers (int) – The maximum number of workers to use for the partition step. Defaults to 256.
  • overview_engine (str) – The engine to use for the overview step. By default, tries first with "realtime" and then falls back to "large".
  • overview_max_workers (int) – The maximum number of workers to use for the overview step. Defaults to 256.

The extract, partition and overview steps are run in parallel using fused.submit(). By default, the function will first attempt to run this using "realtime" instances, and retry any failed runs using "large" instances.

You can override this behavior by specifying the engine, max_workers, worker_concurrency, etc parameters as additional keyword arguments to this function (**kwargs). If you want to specify those per step, use extract_engine, partition_engine, etc. For example, to run everything locally on the same machine where this function runs, use:

run_ingest_raster_to_h3(..., engine="local")

set_visibility​

set_visibility(
    path: str, visibility: str, *, base_url: str | None = None
) -> dict[str, Any]

Make your team's H3 index at path "team" (readable by your team only) or "public" (readable by every Fused user; still re-indexed and deleted only by your team). A public location's files must be publicly readable on S3, so any UDF can read them.

Parameters:

  • path (str) – The indexed location.
  • visibility (str) – "team" or "public".
  • base_url (str | None) – Base URL for the API. If None, uses the current environment.

Returns:

  • dict[str, Any] – The index summary, as fused.h3.index_status.

Raises:

  • HTTPError – 404 if your team has no index at path; 403 when making it public as anyone but a Fused admin; 400 when making it public while its files are not publicly readable on S3.

Run​

A run of a task over several jobs, such as fused.h3.partition.

The server launches one job per job id in a single call and keeps a record of the group under id (it starts with grp:), so fused.h3.get_run(id) brings the run back in a new session. id, task, engine and info never change and are plain attributes.

The rest is read live: jobs() (job id to status), status() (one answer for the run: "failed" if any job failed, else "succeeded" if all succeeded, else "running" if any is running, else "pending"), done() (every job has finished, some may have failed), errors() (job id to error for the failed jobs) and the job lists failed(), succeeded(), skipped() and pending(). A job is pending while it waits to start, for instance because the team has no quota left; the job starts by itself once it fits. A job of a fused.h3.partition run over a udf or era5 source is first read by a job of its own and then partitioned: tail() and repr show which. Reads made within a second of each other share one request, and a finished run is final and needs none. repr makes no request.

done​

done() -> bool

Whether every job has finished, successfully or not.

errors​

errors() -> dict[str, str | None]

The error of each failed job, by job id.

failed​

failed() -> list[str]

Ids of the jobs that failed.

jobs​

jobs() -> dict[str, str | None]

Each job's status by job id: "pending", "running", "succeeded" or "failed".

parts​

parts() -> dict[str, str | None]

Deprecated: use jobs().

pending​

pending() -> list[str]

Ids of the jobs waiting to start.

result​

result() -> Any

The run's outcome once it succeeded: its info (the output of a fused.h3.partition run) with the job ids as parts. None while it is still going.

Raises:

  • RuntimeError – A job failed (the message lists them).

retry​

retry(jobs: list[str] | None = None, **deprecated: Any) -> Run

Run the failed jobs again (or only jobs, which must have failed) and return the run. Jobs that succeeded are not touched.

A job that failed while reading is read again; one that failed while partitioning is partitioned again from what was read.

Raises:

  • HTTPError – The server refused: a job is unknown or did not fail, or none failed.

skipped​

skipped() -> list[str]

Ids of the jobs that were already done and left alone (redo=False); only fused.h3.partition skips jobs.

status​

status() -> str

"failed" if any job failed, else "succeeded" if all did, else "running" if any is running, else "pending".

succeeded​

succeeded() -> list[str]

Ids of the jobs that ran to the end and wrote files (not the skipped ones).

tail​

tail(refresh_seconds: float = 2.0, timeout: float | None = None) -> Run

Print every part's new log lines as [job] message until all jobs have finished, a line saying why a job is waiting, and a one-line summary whenever the counts change. Failed jobs' errors come last.

Logs are best effort: a failed log fetch warns once and polling goes on.

Raises:

  • TimeoutError – Jobs were still going after timeout seconds. They keep running on the server.

wait​

wait(refresh_seconds: float = 2.0, timeout: float | None = None) -> Run

tail() without printing anything.

sources.era5​

era5(
    *,
    outputs: Sequence[str],
    month: str | None = None,
    date: str | None = None,
    cadence: str = "daily",
    bbox: Sequence[float] | None = None,
    read_workers: int = 16
) -> Source

Read a month or a day of ERA5 weather from the public ARCO store as lat/lng rows.

ARCO ERA5 is Google's public copy of the ECMWF ERA5 reanalysis: hourly global weather on a 0.25 degree latitude/longitude raster, 1940 to now. Reading needs network access to storage.googleapis.com. Decoding the zarr chunks needs numcodecs, which the Fused runtime provides; anywhere else, install it yourself.

The time range is the source's input: exactly one of month and date. Give it here to read that range in one pass, or leave it out and give one per part to fused.h3.partition(..., partition_by=[{"month": "2024-01"}, {"month": "2024-02"}]).

Yields record batches with columns lat, lng, month (a string such as "2024-01"), date (date32), hour (uint8, hourly only) and one float32 column per output, in the order given. Longitudes are converted to [-180, 180) for a tidy output column; the nearest-point fill does not need it. Cells are not computed here: hex it with fused.h3.partition(..., point_column=("lat", "lng"), fill="nearest"), which fills every cell from its nearest raster point.

A daily call yields the range one lat/lng tile at a time (90 by 90 degrees, so 8 tiles over the globe). A tile batch holds every requested day, day-major: one complete raster of the tile's 0.25 degree points per date. Tiles are disjoint sets of raster points, so each cell is filled from exactly one tile. Without partition_by, partition therefore writes one monthly file per H3 parent cell directly; only parents that straddle a tile edge are hit by two tiles and merged. partition_by="date" still works and writes one folder per day instead.

An hourly call yields one batch per day (a month of hourly rasters is about 3 GB per column, too much to hold at once). A batch holds all 24 hours, hour-major: 24 complete rasters, one per hour 0 to 23. Pass partition_by="date" so each day becomes a folder (output/date=<date>/<parent>.parquet) of one file per H3 parent cell.

This function is the template for a new datasource: any generator that yields batches of this shape (a complete regular lat/lng raster per key tuple, plus key and value columns) works with partition. partition treats every column that is not floating point (date, hour) as a key and fills the float32 outputs.

Parameters:

  • outputs (Sequence[str]) – Output columns to read. Names depend on cadence: daily has statistics such as t2m_min, t2m_max, t2m_mean, tp (precipitation, mm) and ssrd (radiation, W/m2); hourly has one column per variable such as t2m and tp.
  • month (str | None) – Input. One month as "YYYY-MM".
  • date (str | None) – Input. One day as "YYYY-MM-DD". Give this or month.
  • cadence (str) – "daily" reduces each day's hours to one row per point per day and yields the range in lat/lng tiles. "hourly" keeps every hour (24 times the rows) in one batch per day and reads one day at a time to bound memory.
  • bbox (Sequence[float] | None) – (min_lng, min_lat, max_lng, max_lat) in degrees, longitude in [-180, 180], to keep only the raster points inside the box (edges inclusive). min_lng > max_lng means the box crosses the antimeridian. Default None reads the whole globe. The store's chunks hold the whole globe per hour, so bbox saves memory and output, not download. Cells emitted by partition are those whose nearest raster point is in the box. The box must keep at least two latitude rows and two longitude columns (span at least two 0.25 degree points on each axis).
  • read_workers (int) – Concurrent chunk downloads.

Raises:

  • ValueError – Unknown cadence or outputs, a month or date that is not valid text of its form, both of them given, or a bbox that is malformed, out of range, selects no raster point, or spans fewer than two points on an axis. Iterating without either of month and date raises too.
  • Era5SourceError – Raised during iteration, not at the call, when the ARCO store cannot be read or numcodecs is missing.

Example:

from fused import h3
h3.partition(
    h3.sources.era5(outputs=["t2m_mean", "tp"]),
    "fd://me/era5/",
    partition_by=[{"month": "2024-01"}, {"month": "2024-02"}],
    point_column=("lat", "lng"), fill="nearest", h3_res=7,
    file_res=1, read_engine="large",
)  # doctest: +SKIP

# Hourly takes the same call with a ``date`` per part and
# ``partition_by=["date"]``; each day's file holds all 24 hours.

h3.partition(
    h3.sources.era5(outputs=["t2m"], cadence="hourly"),
    "fd://me/era5_hourly/",
    partition_by=[{"date": "2024-01-01"}, {"date": "2024-01-02"}],
    point_column=("lat", "lng"), fill="nearest", h3_res=7,
    file_res=1,
)  # doctest: +SKIP

sources.parquet​

parquet(
    path: str | Path,
    *,
    columns: Sequence[str] | None = None,
    batch_size: int | None = 65536,
    **hive_keys: Any
) -> Source

Stream a parquet file, or every .parquet file under a folder.

Files are read in sorted path order. path can be a local path or any fsspec URL (fd://, file://, s3://, gs://, az://, memory://). Remote schemes need their fsspec backend installed (s3fs, gcsfs, adlfs).

Parameters:

  • path (str | Path) – A parquet file or a folder of them (searched recursively).
  • columns (Sequence[str] | None) – Read only these columns. Default is all columns.
  • batch_size (int | None) – Maximum rows per yielded batch. None yields each parquet row group as one batch, however large.
  • **hive_keys (Any) – The subfolder to read, as hive keys: year=2020 reads <path>/year=2020/, and several keys nest in the order given (year=2020, month=1 reads <path>/year=2020/month=1/). Each key is added to every batch as a constant string column when the files lack it (hive writers leave it out of the files). With fused.h3.partition(..., partition_by=[{"year": 2020}, ...]) each dict is one part. path, columns and batch_size cannot be hive keys.

Nothing is listed or opened until the first batch is read, so a source over remote storage can be described (and sent to the Fused server) without storage credentials.

Raises:

  • ValueError – A hive key or value is not a plain name or scalar. Also raised on the first iteration, not at the call, when no parquet files were found.

sources.table​

table(data: Any) -> Iterator[pa.RecordBatch]

Yield the batches of a pandas.DataFrame, pyarrow.Table or pyarrow.RecordBatch.

A DataFrame index becomes a column only when every level is named; otherwise (a RangeIndex or an unnamed index, such as after filtering) it is dropped. A geopandas.GeoDataFrame's geometry columns arrive as WKB (binary), the form fused.h3.partition(..., point_column=...) expects; GeoParquet read with parquet is already WKB. The centroid is planar in the coordinates as stored (lon/lat degrees expected), so geometries that cross the antimeridian or wrap a pole are mis-hexed; split them or pass lat/lng columns as point_column= for those rows.

Raises:

  • TypeError – data is none of those.

sources.udf​

udf(udf: Any) -> Source

A source that runs your UDF on the engine, once per part.

The UDF is the reader you would have written: it takes one part's inputs (its parameters) and returns the part's rows. fused.h3.partition runs it in a batch job of its own and writes the rows as parquet for the partition job, so nothing is read from this machine::

@fused.udf
def read_month(month: int):
    return load_ais(month)  # a DataFrame, GeoDataFrame, Table or RecordBatch

fused.h3.partition(
    fused.h3.sources.udf(read_month),
    "fd://me/ais/",
    partition_by={"month": range(1, 13)},
    point_column="geometry",
)

Parameters:

  • udf (Any) – A UDF made with @fused.udf. Its code is serialized and run on the engine, so it must be self-contained: import what it needs inside the body, and do not rely on files or variables that exist only on this machine. The UDF's parameters are the source's inputs: a parameter with no default is required. Pass them to Source.bind or as one dict per part in parts=. With parts, each dict's keys are output columns and become the partition columns: the UDF must return those columns, with those values, in every row. Without parts the UDF runs once, with its defaults and any inputs given as Source.bind.

The UDF must return a pandas DataFrame, a GeoDataFrame, a pyarrow Table or a RecordBatch. Anything else, a generator included, raises TypeError in the run.

Iterating a udf source here runs the UDF in this process, which fused.h3.partition never does.

Raises:

  • TypeError – udf is not a fused.models.udf.Udf (a plain function or a string is not accepted).