Iterate through datapoints in chunks

async AsyncCogniteClient.time_series.data.__call__(
queries: DatapointsQuery | Sequence[DatapointsQuery],
*,
chunk_size_datapoints: int = 100000,
chunk_size_time_series: int | None = None,
return_arrays: bool = True,
) → AsyncIterator[DatapointsArray | DatapointsArrayList | Datapoints | DatapointsList]

Iterate through datapoints in chunks, for one or more time series.

Note

Control memory usage by specifying chunk_size_time_series, how many time series to iterate simultaneously and chunk_size_datapoints, how many datapoints to yield per iteration (per individual time series). See full example in examples. Note that in order to make efficient use of the API request limits, this method will never hold less than 100k datapoints in memory at a time, per time series.

If you run with memory constraints, use return_arrays=True (the default).

No empty chunk is ever returned.

Parameters:

  • queries (DatapointsQuery | Sequence[DatapointsQuery]) – Query, or queries, using id, external_id or instance_id for the time series to fetch data for, with individual settings specified. The options ‘limit’ and ‘include_outside_points’ are not supported when iterating.

  • chunk_size_datapoints (int) – The number of datapoints per time series to yield per iteration. Must evenly divide 100k OR be an integer multiple of 100k. Default: 100_000.

  • chunk_size_time_series (int | None) – The max number of time series to yield per iteration (varies as time series get exhausted, but is never empty). Default: None (all given queries are iterated at the same time).

  • return_arrays (bool) – Whether to return the datapoints as numpy arrays. Default: True.

Yields:

DatapointsArray | DatapointsArrayList | Datapoints | DatapointsList – If return_arrays=True, a DatapointsArray object containing the datapoints chunk, or a DatapointsArrayList if multiple time series were asked for. When False, a Datapoints object containing the datapoints chunk, or a DatapointsList if multiple time series were asked for.

Examples

Iterate through the datapoints of a single time series with external_id=”foo”, in chunks of 25k:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import DatapointsQuery
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> query = DatapointsQuery(external_id="foo", start="2w-ago")
>>> for chunk in client.time_series.data(query, chunk_size_datapoints=25_000):
...     pass  # do something with the datapoints chunk

Iterate through datapoints from multiple time series, and do not return them as memory-efficient numpy arrays. As one or more time series get exhausted (no more data), they are no longer part of the returned “chunk list”. Note that the order is still preserved (for the remaining).

If you run with chunk_size_time_series=None, an easy way to check when a time series is exhausted is to use the .get method, as illustrated below:

>>> from cognite.client.data_classes.data_modeling import NodeId
>>> queries = [
...     DatapointsQuery(id=123),
...     DatapointsQuery(external_id="foo"),
...     DatapointsQuery(instance_id=NodeId("my-space", "my-ts-xid")),
... ]
>>> for chunk_lst in client.time_series.data(query, return_arrays=False):
...     if chunk_lst.get(id=123) is None:
...         print("Time series with id=123 has no more datapoints!")

A likely use case for iterating datapoints is to clone data from one project to another, while keeping a low memory footprint and without having to write very custom logic involving count aggregates (which won’t work for string data) or do time-domain splitting yourself.

Here’s an example of how to do so efficiently, while including bad- and uncertain data (ignore_bad_datapoints=False) and copying status codes (include_status=True). This is automatically taken care of when the Datapoints(-Array) objects are passed directly to an insert method. The only assumption below is that the time series have already been created in the target project.

>>> from cognite.client.utils import MIN_TIMESTAMP_MS, MAX_TIMESTAMP_MS
>>> target_client = CogniteClient()
>>> ts_to_copy = client.time_series.list(data_set_external_ids="my-use-case")
>>> queries = [
...     DatapointsQuery(
...         external_id=ts.external_id,
...         include_status=True,
...         ignore_bad_datapoints=False,
...         start=MIN_TIMESTAMP_MS,
...         end=MAX_TIMESTAMP_MS + 1,  # end is exclusive
...     )
...     for ts in ts_to_copy
... ]
>>> for dps_chunk in client.time_series.data(
...     queries,  # may be several thousand time series...
...     chunk_size_time_series=20,  # control memory usage by specifying how many to iterate at a time
...     chunk_size_datapoints=100_000,
... ):
...     target_client.time_series.data.insert_multiple(
...         [{"external_id": dps.external_id, "datapoints": dps} for dps in dps_chunk]
...     )