awswrangler.timestream.write

awswrangler.timestream.write(df: DataFrame, database: str, table: str, time_col: str | None = None, measure_col: str | List[str | None] | None = None, dimensions_cols: List[str | None] | None = None, version: int = 1, use_threads: bool | int = True, measure_name: str | None = None, common_attributes: Dict[str, Any] | None = None, boto3_session: Session | None = None) → List[Dict[str, str]]

Store a Pandas DataFrame into an Amazon Timestream table.

Note

In case use_threads=True, the number of threads from os.cpu_count() is used.

If the Timestream service rejects a record(s), this function will not throw a Python exception. Instead it will return the rejection information.

Note

If time_col column is supplied, it must be of type timestamp. TimeUnit is set to MILLISECONDS by default.

Note

Following arguments are not supported in distributed mode with engine EngineEnum.RAY:

  • boto3_session

Parameters:

  • df (pandas.DataFrame) – Pandas DataFrame https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html

  • database (str) – Amazon Timestream database name.

  • table (str) – Amazon Timestream table name.

  • time_col (Optional[str]) – DataFrame column name to be used as time. MUST be a timestamp column.

  • measure_col (Union[str, List[str], None]) – DataFrame column name(s) to be used as measure.

  • dimensions_cols (Optional[List[str]]) – List of DataFrame column names to be used as dimensions.

  • version (int) – Version number used for upserts. Documentation https://docs.aws.amazon.com/timestream/latest/developerguide/API_WriteRecords.html.

  • use_threads (bool, int) – True to enable concurrent writing, False to disable multiple threads. If enabled, os.cpu_count() is used as the number of threads. If integer is provided, specified number is used.

  • measure_name (Optional[str]) – Name that represents the data attribute of the time series. Overrides measure_col if specified.

  • common_attributes (Optional[Dict[str, Any]]) – Dictionary of attributes shared across all records in the request. Using common attributes can optimize the cost of writes by reducing the size of request payloads. Values in common_attributes take precedence over all other arguments and data frame values. Dimension attributes are merged with attributes in record objects. Example: {"Dimensions": [{"Name": "device_id", "Value": "12345"}], "MeasureValueType": "DOUBLE"}.

  • boto3_session (boto3.Session(), optional) – Boto3 Session. If None, the default boto3 Session is used.

Returns:

Rejected records. Possible reasons for rejection are described here: https://docs.aws.amazon.com/timestream/latest/developerguide/API_RejectedRecord.html

Return type:

List[Dict[str, str]]

Examples

Store a Pandas DataFrame into a Amazon Timestream table.

>>> import awswrangler as wr
>>> import pandas as pd
>>> df = pd.DataFrame(
>>>     {
>>>         "time": [datetime.now(), datetime.now(), datetime.now()],
>>>         "dim0": ["foo", "boo", "bar"],
>>>         "dim1": [1, 2, 3],
>>>         "measure": [1.0, 1.1, 1.2],
>>>     }
>>> )
>>> rejected_records = wr.timestream.write(
>>>     df=df,
>>>     database="sampleDB",
>>>     table="sampleTable",
>>>     time_col="time",
>>>     measure_col="measure",
>>>     dimensions_cols=["dim0", "dim1"],
>>> )
>>> assert len(rejected_records) == 0

Return value if some records are rejected.

>>> [
>>>     {
>>>         'ExistingVersion': 2,
>>>         'Reason': 'The record version 1 is lower than the existing version 2. A '
>>>                   'higher version is required to update the measure value.',
>>>         'RecordIndex': 0
>>>     }
>>> ]