有关与适用于 LiveAnalytics 的 Amazon Timestream 类似的功能，可以考虑使用适用于 InfluxDB 的 Amazon Timestream。适用于 InfluxDB 的 Amazon Timestream 提供简化的数据摄取和个位数毫秒级的查询响应时间，以实现实时分析。点击此处了解更多信息。

读取

您可以从联网设备、IT 系统和工业设备收集时间序列数据，并将其写入适用于 LiveAnalytics 的 Timestream。当时间序列属于同一个表时，适用于 LiveAnalytics 的 Timestream 支持在单个写入请求中写入来自单个时间序列的数据点和/或来自多个序列的数据点。为方便起见，适用于 LiveAnalytics 的 Timestream 提供灵活的架构，能够根据您在调用数据库写入操作时指定的度量值的维度名称和数据类型，自动检测适用于 LiveAnalytics 的 Timestream 表的列名和数据类型。您也可以将批量数据写入适用于 LiveAnalytics 的 Timestream。

您可以使用 AWS SDK、AWS CLI 写入数据，也可以通过 AWS Lambda、AWS IoT Core、适用于 Apache Flink 的亚马逊托管服务、Amazon Kinesis、Amazon MSK 和 开源 Telegraf 写入数据。

数据类型

适用于 LiveAnalytics 的 Timestream 支持以下数据类型进行写入。

无预先架构定义

在将数据发送到适用于 LiveAnalytics 的 Amazon Timestream 之前，您必须使用 AWS 管理控制台、适用于 LiveAnalytics 的 Timestream SDK 或适用于 LiveAnalytics 的 Timestream API 操作创建数据库和表。有关更多信息，请参阅创建数据库和创建表。创建表时，无需预先定义架构。适用于 LiveAnalytics 的 Amazon Timestream 能够根据发送数据点的度量和维度自动检测架构，因此您无需再为适应快速变化的时间序列数据而离线修改架构。

写入数据（插入和更新插入）

适用于 LiveAnalytics 的 Amazon Timestream 中的写入操作使您能够插入和更新插入数据。默认情况下，适用于 LiveAnalytics 的 Amazon Timestream 中的写入遵循首位写入者胜出语义，即数据仅存储为附加，重复记录将被拒绝。虽然首位写入者胜出语义可满足许多时间序列应用程序的需求，但在某些场景下，应用程序需要以幂等方式更新现有记录和/或采用最后一位写入者胜出语义写入数据，即服务端存储版本最高的记录。为解决这些情况，适用于 LiveAnalytics 的 Amazon Timestream 提供更新插入数据的功能。更新插入是一种操作，当记录不存在时将其插入系统，当记录存在时则更新该记录。当记录更新时，其更新过程具有幂等性。

删除操作不包含记录级别。但可以删除表和数据库。

将数据写入内存存储和磁性存储

适用于 LiveAnalytics 的 Amazon Timestream 能够将数据直接写入内存存储和磁性存储。内存存储针对高吞吐量数据写入进行优化，磁性存储针对延迟到达数据的低吞吐量写入进行优化。

延迟到达的数据是指时间戳早于当前时间且超出内存存储保留期的数据。您必须通过为该表启用磁性存储写入功能，明确允许将延迟到达的数据写入磁性存储。此外，MagneticStoreRejectedDataLocation 在创建表时进行定义。要写入磁性存储，WriteRecords 的调用者必须在创建表期间对 MagneticStoreRejectedDataLocation 中指定的 S3 存储桶具备 S3:PutObject 权限。有关更多信息，请参阅 CreateTable、WriteRecords 和 PutObject。

使用单度量记录和多度量记录写入数据

适用于 LiveAnalytics 的 Amazon Timestream 支持使用两种类型的记录写入数据，即单度量记录和多度量记录。

单度量记录

单度量记录允许您每条记录发送一个度量。当使用这种格式将数据发送到适用于 LiveAnalytics 的 Timestream 时，适用于 LiveAnalytics 的 Timestream 会为每条记录创建一个表行。这意味着，如果某个设备发出 4 个指标，且每个指标都作为单度量记录发送，则适用于 LiveAnalytics 的 Timestream 将在表中创建 4 个行以存储这些数据，并且每行都具有相同的设备属性。如果需要监控应用程序中的单个指标，或者应用程序不会同时发出多个指标，则建议使用此格式。

多度量记录

使用多度量记录时，您可以在单个表行中存储多个度量，而非每个表行存储一个度量。因此，多度量记录使您能够将现有数据从关系数据库迁移至适用于 LiveAnalytics 的 Amazon Timestream，且仅需极少修改。

您还可以在单次写入请求中批量处理比单度量记录更多的数据。这可提高数据写入吞吐量和性能，还可降低数据写入成本。这是因为在写入请求中批量处理更多数据，使适用于 LiveAnalytics 的 Amazon Timestream 能够在单次写入请求中识别更多可重复数据（如适用），并对重复数据仅收取一次费用。

多度量记录

借助多度量记录，您可以在内存存储和磁性存储器中以更紧凑的格式存储时间序列数据，这有助于降低数据存储成本。此外，紧凑的数据存储有助于写入更简单的查询，以进行数据检索，提升查询性能并降低查询成本。

此外，多度量记录还支持 TIMESTAMP 数据类型，可用于在时间序列记录中存储多个时间戳。多度量记录中，TIMESTAMP 属性支持未来或过去的时间戳。因此，多度量记录有助于提升性能、降低成本并简化查询，同时为存储不同类型的关联度量提供更大的灵活性。

示例：监控视频流媒体应用程序的性能和运行状况

假设在 200 个 EC2 实例上运行的视频流媒体应用程序。您希望使用适用于 LiveAnalytics 的 Amazon Timestream 存储和分析应用程序发出的指标，以便了解应用程序的性能和运行状况，快速识别异常情况，解决问题并发现优化机会。

我们将分别使用单度量记录和多度量记录对此场景进行建模，然后比较/对比这两种方法。针对每种方法，我们作出以下假设。

使用单度量记录

数据建模：使用单度量记录，我们将针对四个度量值（视频启动时间、重新缓冲比率、视频播放失败和平均帧速率）分别创建一条记录。每条记录将包含四个维度（device_id、device_type、os_version 和 region）和一个时间戳。

写入：将数据写入适用于 LiveAnalytics 的 Amazon Timestream 时，记录的构造方式如下。

public void writeRecords() {
    System.out.println("Writing records");
    // Specify repeated values for all records
    List<Record> records = new ArrayList<>();
    final long time = System.currentTimeMillis();
 
    List<Dimension> dimensions = new ArrayList<>();
    
    final Dimension device_id = new Dimension().withName("device_id").withValue("12345678");
    final Dimension device_type = new Dimension().withName("device_type").withValue("iPhone 11");
    final Dimension os_version = new Dimension().withName("os_version").withValue("14.8");
    final Dimension region = new Dimension().withName("region").withValue("us-east-1");
 
    dimensions.add(device_id);
    dimensions.add(device_type);
    dimensions.add(os_version);
    dimensions.add(region);
 
    Record videoStartupTime = new Record()
        .withDimensions(dimensions)
        .withMeasureName("video_startup_time")
        .withMeasureValue("200")
        .withMeasureValueType(MeasureValueType.BIGINT)
        .withTime(String.valueOf(time));
    Record rebufferingRatio = new Record()
        .withDimensions(dimensions)
        .withMeasureName("rebuffering_ratio")
        .withMeasureValue("0.5")
        .withMeasureValueType(MeasureValueType.DOUBLE)
        .withTime(String.valueOf(time));
    Record videoPlaybackFailures = new Record()
        .withDimensions(dimensions)
        .withMeasureName("video_playback_failures")
        .withMeasureValue("0")
        .withMeasureValueType(MeasureValueType.BIGINT)
        .withTime(String.valueOf(time));
    Record averageFrameRate = new Record()
        .withDimensions(dimensions)
        .withMeasureName("average_frame_rate")
        .withMeasureValue("0.5")
        .withMeasureValueType(MeasureValueType.DOUBLE)
        .withTime(String.valueOf(time));

    records.add(videoStartupTime);
    records.add(rebufferingRatio); 
    records.add(videoPlaybackFailures);
    records.add(averageFrameRate);
 
    WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
        .withDatabaseName(DATABASE_NAME)
        .withTableName(TABLE_NAME)
        .withRecords(records);
 
    try {
      WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
      System.out.println("WriteRecords Status: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
    } catch (RejectedRecordsException e) {
      System.out.println("RejectedRecords: " + e);
      for (RejectedRecord rejectedRecord : e.getRejectedRecords()) {
        System.out.println("Rejected Index " + rejectedRecord.getRecordIndex() + ": "
            + rejectedRecord.getReason());
      }
      System.out.println("Other records were written successfully. ");
    } catch (Exception e) {
      System.out.println("Error: " + e);
    }
  }

存储单度量记录时，数据的逻辑表示方式如下。

查询：您可以写入查询，以检索过去 15 分钟内收到的所有具有相同时间戳的数据点，如下所示。

with cte_video_startup_time as ( SELECT time, device_id, device_type, os_version, region, measure_value::bigint as video_startup_time FROM table where time >= ago(15m) and measure_name=”video_startup_time”),
cte_rebuffering_ratio as ( SELECT time, device_id, device_type, os_version, region, measure_value::double as rebuffering_ratio FROM table where time >= ago(15m) and measure_name=”rebuffering_ratio”),
cte_video_playback_failures as ( SELECT time, device_id, device_type, os_version, region, measure_value::bigint as video_playback_failures FROM table where time >= ago(15m) and measure_name=”video_playback_failures”),
cte_average_frame_rate as ( SELECT time, device_id, device_type, os_version, region, measure_value::double as average_frame_rate FROM table where time >= ago(15m) and measure_name=”average_frame_rate”)
SELECT a.time, a.device_id, a.os_version, a.region, a.video_startup_time, b.rebuffering_ratio, c.video_playback_failures, d.average_frame_rate FROM cte_video_startup_time a, cte_buffering_ratio b, cte_video_playback_failures c, cte_average_frame_rate d WHERE
a.time = b.time AND a.device_id = b.device_id AND a.os_version = b.os_version AND a.region=b.region AND
a.time = c.time AND a.device_id = c.device_id AND a.os_version = c.os_version AND a.region=c.region AND
a.time = d.time AND a.device_id = d.device_id AND a.os_version = d.os_version AND a.region=d.region

工作负载成本：该工作负载的成本估算为每月 373.23 美元（基于单度量记录）。

使用多度量记录

数据建模：对于多度量记录，我们将创建一条记录，包含所有四个度量（视频启动时间、重新缓冲比率、视频播放失败率和平均帧速率）、所有四个维度（device_id、deversion_type、os_version 和 region）以及一个时间戳。

写入：将数据写入适用于 LiveAnalytics 的 Amazon Timestream 时，记录的构造方式如下。

public void writeRecords() {
    System.out.println("Writing records");
    // Specify repeated values for all records
    List<Record> records = new ArrayList<>();
    final long time = System.currentTimeMillis();
 
    List<Dimension> dimensions = new ArrayList<>();
    
    final Dimension device_id = new Dimension().withName("device_id").withValue("12345678");
    final Dimension device_type = new Dimension().withName("device_type").withValue("iPhone 11");
    final Dimension os_version = new Dimension().withName("os_version").withValue("14.8");
    final Dimension region = new Dimension().withName("region").withValue("us-east-1");
 
    dimensions.add(device_id);
    dimensions.add(device_type);
    dimensions.add(os_version);
    dimensions.add(region);
 
    Record videoMetrics = new Record()
        .withDimensions(dimensions)
        .withMeasureName("video_metrics")
        .withTime(String.valueOf(time));
        .withMeasureValueType(MeasureValueType.MULTI)
        .withMeasureValues(
          new MeasureValue()
        	.withName("video_startup_time")
        	.withValue("0")
        	.withValueType(MeasureValueType.BIGINT),
        	new MeasureValue()
		.withName("rebuffering_ratio")
        	.withValue("0.5")
        	.withType(MeasureValueType.DOUBLE),
          new MeasureValue()
        	.withName("video_playback_failures")
        	.withValue("0")
        	.withValueType(MeasureValueType.BIGINT),
		 new MeasureValue()
         	.withName("average_frame_rate")
        	.withValue("0.5")
        	.withValueType(MeasureValueType.DOUBLE))
 
    records.add(videoMetrics);
 
    WriteRecordsRequest writeRecordsRequest = new WriteRecordsRequest()
        .withDatabaseName(DATABASE_NAME)
        .withTableName(TABLE_NAME)
        .withRecords(records);
 
    try {
      WriteRecordsResult writeRecordsResult = amazonTimestreamWrite.writeRecords(writeRecordsRequest);
      System.out.println("WriteRecords Status: " + writeRecordsResult.getSdkHttpMetadata().getHttpStatusCode());
    } catch (RejectedRecordsException e) {
      System.out.println("RejectedRecords: " + e);
      for (RejectedRecord rejectedRecord : e.getRejectedRecords()) {
        System.out.println("Rejected Index " + rejectedRecord.getRecordIndex() + ": "
            + rejectedRecord.getReason());
      }
      System.out.println("Other records were written successfully. ");
    } catch (Exception e) {
      System.out.println("Error: " + e);
    }
  }

存储多度量记录时，数据的逻辑表示方式如下。

查询：您可以写入查询，以检索过去 15 分钟内收到的所有具有相同时间戳的数据点，如下所示。

SELECT time, device_id, device_type, os_version, region, video_startup_time, rebuffering_ratio, video_playback_failures, average_frame_rate FROM table where time >= ago(15m)

工作负载成本：该工作负载的成本估算为 127.43 美元（基于多度量记录）。

使用过去或将来的时间戳写入数据

适用于 LiveAnalytics 的 Timestream 通过多种机制，支持将带有超出内存存储保留窗口时间戳的数据写入系统。

读取的最终一致性

适用于 LiveAnalytics 的 Timestream 支持读取的最终一致性语义。这意味着，如果您将批量数据写入适用于 LiveAnalytics 的 Timestream 后立即查询数据，查询结果可能不会反映最近完成的写入操作结果。如果您在短时间后重复这些读取请求，结果将返回最新的数据。