Time-series

Time-series metadata, datapoint retrieval, and datapoint ingestion (single-request or high-throughput).

A series' externalId identifies it: unique per tenant, compared without case, and stored exactly as you send it. External ids & naming →

Create a series

Java

Timeseries series = new Timeseries()
        .setExternalId("engine_temperature")
        .setName("Engine temperature");
series.setUnit("celsius");

client.timeseries().create(List.of(series));

Python

import datahub_sdk

ts = datahub_sdk.TimeSeries(
    external_id="engine_temperature",
    name="Engine temperature",
    unit="celsius")

client.timeseries.create([ts])

Rust

use dataplatform_rust_sdk::timeseries::TimeSeries;

let mut ts = TimeSeries::new("engine_temperature", "Engine temperature");
ts.unit = Some("celsius".into());
api.time_series.create_one(&ts).await?;

Value types

Every series has a value type that decides how its datapoints are stored. Leave it unset and the series is floating-point (float32) — right for most sensor readings, so the create above accepts decimal values as-is. Set it explicitly when you need something else:

Value type	Use it for
float32 (default)	Sensor readings — 32-bit precision is plenty.
float	Double-precision floating point.
numeric / decimal32	Exact decimals — money, lab values — stored without floating-point rounding. Pass the values as strings.
bigint	Whole numbers (counts, integer statuses).
text	Non-numeric string values.
mixed	Heterogeneous values in one series.

A float written to a bigint series is rejected, so pick the type that matches the data. For a value that must reconcile exactly, use numeric:

Java

Timeseries price = new Timeseries()
        .setExternalId("book_value_usd")
        .setName("Book value (USD)");
price.setUnit("usd");
price.setValueType("numeric");        // exact decimals, no float rounding

client.timeseries().create(List.of(price));

Python

client.timeseries.create([datahub_sdk.TimeSeries(
    external_id="book_value_usd", name="Book value (USD)",
    unit="usd", value_type="numeric")])

Rust

let mut price = TimeSeries::new("book_value_usd", "Book value (USD)");
price.unit = Some("usd".into());
price.value_type = "numeric".into();  // exact decimals, no float rounding
api.time_series.create_one(&price).await?;

Filter series

POST /timeseries/filter finds series by structured criteria. Everything you supply is combined with AND — a series must match every criterion to be included.

Criterion	Matching
dataSetId	The data set and every data set beneath it in the data set hierarchy — a series attached to a child (or grandchild, …) data set matches too.
unit	Case-insensitive; % works as a wildcard ("cel%").
unitExternalId	Exact match on the unit-catalogue external id (e.g. temperature_deg_c).
metadataKey / metadataValue	Together: that key must carry that value. Alone: any entry with that key (or any entry with that value).

Results come newest first, capped by limit (default 1000, max 10000). Series in data sets you lack read access to are silently omitted — the result is what your token may see, not an error. For free-text lookups use POST /timeseries/search instead.

Java

import ai.intellistream.datahub.models.datafilters.TimeseriesFilter;

TimeseriesFilter criteria = new TimeseriesFilter();
criteria.setDataSetId(12L);      // this data set and every data set beneath it
criteria.setUnit("celsius");

DataWrapper<Timeseries> series = client.timeseries().filter(criteria);

Pass a TimeseriesRetreiver instead of the bare criteria to set an explicit limit.

Python

form = datahub_sdk.TimeSeriesFilterForm(
    data_set_id=12,              # this data set and every data set beneath it
    unit="celsius",
    limit=100)

series = client.timeseries.filter(form)

Rust

use dataplatform_rust_sdk::{TimeSeriesFilter, TimeSeriesFilterForm};

let criteria = TimeSeriesFilter {
    data_set_id: Some(12),       // this data set and every data set beneath it
    unit: Some("celsius".into()),
    ..Default::default()
};
let series = api.time_series.filter(&TimeSeriesFilterForm::new(criteria, Some(100))).await?;

The hierarchy expansion is what makes "master" data sets useful: filter on the top-level data set of a site or project and you get the series of the whole family beneath it, without knowing (or maintaining a list of) the sub-data sets.

Delete a series

Deletes the series and its datapoints. Remove any referencing subscriptions (and edges) first, or the backend responds 409.

Java

import ai.intellistream.datahub.models.IdCollection;

client.timeseries().delete(List.of(IdCollection.createFromExternalId("engine_temperature")));

Python

client.timeseries.delete(["engine_temperature"])

Rust

use dataplatform_rust_sdk::generic::{DataWrapper, IdAndExtId};

api.time_series
    .delete(&DataWrapper::from_vec(vec![IdAndExtId::from_external_id("engine_temperature")]))
    .await?;

Write datapoints

A datapoint is a (timestamp, value) pair grouped under a series' external id.

Java

Timestamps are epoch milliseconds as strings:

DatapointsCollection collection = new DatapointsCollection();
collection.setExternalId("engine_temperature");
collection.setDatapoints(List.of(
        new DatapointString(String.valueOf(System.currentTimeMillis()), "92.4")));

client.timeseries().insertDatapoints(List.of(collection));

Python

Pass timezone-aware timestamps (pandas or datetime); the SDK converts to UTC:

import pandas as pd

client.timeseries.insert_from_lists(
    timestamps=pd.date_range("2026-01-01", periods=3, freq="h", tz="UTC"),
    values=[92.4, 92.6, 92.1],
    ts=ts)

Rust

use chrono::Utc;

api.time_series
    .insert_datapoint(None, Some("engine_temperature".into()), Utc::now(), "92.4".into())
    .await?;

High-throughput ingestion

For large or unbounded volumes the SDK chunks and sends in bulk. See the ingestion guide for the full story.

Survive outages with durable buffering

Enable durable buffering on the client and datapoint ingestion that can't reach the API spools to disk and flushes on the next call, bounded by a time and/or size window. Retries are idempotent (datapoints dedup on (series, timestamp)).

Java

ingest chunks, parallelises and retries, returning an IngestResult tuned with IngestOptions:

IngestResult result = client.timeseries().ingest(data,
        IngestOptions.builder()
                .batchSize(10_000)   // datapoints per request
                .parallelism(16)     // concurrent in-flight requests
                .maxRetries(3)
                .build());

System.out.printf("ingested %,d, failed %,d%n", result.succeeded(), result.failed());

Python

insert_from_lists takes whole arrays (NumPy / pandas) and handles batching for you:

import numpy as np, pandas as pd

client.timeseries.insert_from_lists(
    timestamps=pd.date_range("2026-01-01", periods=1_000_000, freq="s", tz="UTC"),
    values=np.random.rand(1_000_000),
    ts=ts)

Rust

insert_datapoints auto-batches large inputs (chunks above ~100k points):

use dataplatform_rust_sdk::generic::{DataWrapper, DatapointsCollection, DatapointString};

let mut dw = DataWrapper::new();
dw.add_item(DatapointsCollection {
    external_id: Some("engine_temperature".into()),
    datapoints: points,          // Vec<DatapointString { timestamp, value }>
    ..Default::default()
});
api.time_series.insert_datapoints(&mut dw).await?;

Retrieve datapoints

Identify a series (external id or id) and a time window.

Java

import java.time.ZonedDateTime;

RetrieveFilter series = new RetrieveFilter();
series.setExternalId("engine_temperature");
series.setStart(ZonedDateTime.now().minusHours(1));
series.setEnd(ZonedDateTime.now());
series.setLimit(1000);

DataRetriever<RetrieveFilter> request = new DataRetriever<>();
request.setItems(List.of(series));

DataWrapper<DatapointsCollection> points = client.timeseries().retrieve(request);
points.getItems().forEach(c ->
        System.out.println(c.getExternalId() + ": " + c.getDatapoints().size() + " points"));

Python

import pandas as pd

rf = datahub_sdk.RetrieveFilter(
    ts="engine_temperature",
    start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=1),
    end=pd.Timestamp.now(tz="UTC"),
    limit=1000)

collection = client.timeseries.retrieve_datapoints(rf)[0]
for dp in collection.get_datapoints():
    print(dp.timestamp, dp.value)

Rust

use chrono::Utc;
use dataplatform_rust_sdk::generic::{DataWrapper, RetrieveFilter};

let filter = RetrieveFilter {
    external_id: Some("engine_temperature".into()),
    start: Some(Utc::now() - chrono::Duration::hours(1)),
    end: Some(Utc::now()),
    limit: Some(1000),
    ..Default::default()
};
let points = api.time_series
    .retrieve_datapoints(&DataWrapper::from(vec![filter]))
    .await?;
for c in points.get_items() {
    println!("{} points", c.datapoints.len());
}

IngestOptions

The Java ingest tuning knobs (Python's insert_from_lists and Rust's insert_datapoints batch internally):

Option	Default	Meaning
batchSize	10_000	Maximum items per request.
parallelism	8	Concurrent in-flight requests.
maxRetries	3	Retries for transient failures (HTTP 429/5xx, network).
failFast	false	If true, abort on the first failed batch instead of collecting errors.

IngestOptions.defaults() returns the defaults; ingest(data) (no options) uses them.

IngestResult

long              succeeded()    // items ingested
long              failed()       // items that could not be ingested
long              buffered()     // items spooled to the durable buffer (0 unless buffering is on)
boolean           isComplete()   // true when nothing failed and nothing was buffered
List<BatchError>  errors()       // one entry per failed batch

BatchError is a record (int datapointCount, int statusCode, String message) — statusCode is 0 when the failure was a network error rather than an HTTP status.

if (!result.isComplete()) {
    result.errors().forEach(e ->
            System.err.println(e.statusCode() + " on " + e.datapointCount() + " items: " + e.message()));
}