Amazon Timestream for LiveAnalytics に類似した機能をご希望の場合は Amazon Timestream for InfluxDB をご検討ください。リアルタイム分析に適した、シンプルなデータインジェストと 1 桁ミリ秒のクエリ応答時間を特徴としています。詳細については、こちらを参照してください。

writes

接続されたデバイス、IT システム、産業機器から時系列データを収集し、Timestream for LiveAnalytics に書き込むことができます。Timestream for LiveAnalytics では、時系列が同じテーブルに属している場合、単一の時系列からのデータポイントや、単一の書き込みリクエスト内にある多数の系列からのデータポイントを書き込むことができます。Timestream for LiveAnalytics には、データベースへの書き込みを呼び出すときに指定したメジャー値のディメンション名とデータ型に基づいて、Timestream for LiveAnalytics テーブルの列名とデータ型を自動検出する柔軟なスキーマが用意されています。また、データのバッチを Timestream for LiveAnalytics に書き込むこともできます。

AWS SDK、AWS CLI、または AWS Lambda、AWS IoT Core、Amazon Managed Service for Apache Flink、Amazon Kinesis、Amazon MSK、オープンソース Telegraf を使用してデータを書き込むことができます。

データ型

Timestream for LiveAnalytics は、書き込みに対して次のデータ型をサポートしています。

事前スキーマ定義なし

Amazon Timestream for LiveAnalytics にデータを送信する前に、AWS マネジメントコンソール、Timestream for LiveAnalytics SDK または Timestream for LiveAnalytics API オペレーションを使用してデータベースとテーブルを作成する必要があります。詳細については、「データベースを作成する」および「テーブルを作成する」を参照してください。テーブルの作成中に、スキーマを事前に定義する必要はありません。Amazon Timestream for LiveAnalytics は、送信されるデータポイントのメジャーとディメンションに基づいてスキーマを自動的に検出するため、急速に変化する時系列データに合わせてスキーマをオフラインで変更する必要がなくなりました。

データの書き込み (挿入とアップサート)

Amazon Timestream for LiveAnalytics の書き込みオペレーションを使用すると、データを挿入およびアップサートできます。デフォルトでは、Amazon Timestream for LiveAnalytics での書き込みは、first writer wins セマンティクスに従います。このセマンティクスでは、データは追加としてのみ保存され、重複したレコードは拒否されます。first writer wins セマンティクスは、多くの時系列アプリケーションの要件を満たしますが、アプリケーションがべき等性を持つ方法で既存のレコードを更新したり、last writer wins セマンティクスでデータを書き込んだりする必要があるシナリオでは、最も新しいバージョンのレコードがサービスに保存されます。これらのシナリオに対応するため、Amazon Timestream for LiveAnalytics はデータをアップサートする機能を提供します。アップサートは、レコードが存在しない場合はシステムにレコードを挿入し、存在する場合はレコードを更新するオペレーションです。レコードが更新されると、べき等性を持つ方法で更新されます。

削除するためのレコードレベルのオペレーションはありません。ただし、テーブルとデータベースは削除できます。

メモリストアとマグネティックストアへのデータの書き込み

Amazon Timestream for LiveAnalytics では、データをメモリストアとマグネティックストアに直接書き込むことができます。メモリストアは高スループットのデータ書き込みに最適化され、マグネティックストアは遅延到着データの低スループット書き込みに最適化されています。

遅延受信データは、タイムスタンプが現在の時刻より前で、メモリストアの保持期間外のデータです。テーブルのマグネティックストア書き込みを有効にすることで、遅延到着データをマグネティックストアに書き込む機能を明示的に有効にする必要があります。また、MagneticStoreRejectedDataLocation はテーブルの作成時に定義されます。マグネティックストアに書き込むには、WriteRecords の呼び出し元に、テーブルの作成時に MagneticStoreRejectedDataLocation で指定された S3 バケットへの S3:PutObject アクセス許可が必要です。詳細については、「CreateTable」「WriteRecords」「PutObject」を参照してください。

単一メジャーレコードとマルチメジャーレコードを使用したデータの書き込み

Amazon Timestream for LiveAnalytics では、単一メジャーレコードとマルチメジャーレコードの 2 種類のレコードを使用してデータを書き込むことができます。

単一メジャーレコード

単一メジャーレコードを使用すると、レコードごとに 1 つのメジャーを送信できます。この形式を使用してデータを Timestream for LiveAnalytics に送信すると、Timestream for LiveAnalytics はレコードごとに単一のテーブル行を作成します。つまり、デバイスが 4 つのメトリクスを出力し、各メトリクスが単一メジャーレコードとして送信されると、Timestream for LiveAnalytics はテーブルに 4 行を作成してこのデータを保存し、デバイス属性を行ごとに繰り返します。この形式は、アプリケーションの単一のメトリクスをモニタリングする場合や、アプリケーションが同時に複数のメトリクスを出力しない場合に推奨されます。

マルチメジャーレコード

マルチメジャーレコードを使用すると、テーブル行ごとに単一のメジャーを保存するのではなく、複数のメジャーを単一のテーブル行に保存できます。したがって、マルチメジャーレコードを使用すると、最小限の変更で既存のデータをリレーショナルデータベースから Amazon Timestream for LiveAnalytics に移行できます。

単一メジャーレコードよりも多くのデータを 1 回の書き込みリクエストでバッチ処理することもできます。これにより、データ書き込みスループットとパフォーマンスが向上し、データ書き込みのコストも削減されます。なぜなら、書き込みリクエストでより多くのデータをバッチ処理することで、Amazon Timestream for LiveAnalytics は 1 回の書き込みリクエスト (該当する場合) でより多くの繰り返し可能なデータを識別し、繰り返しデータに対して 1 回だけ課金するためです。

マルチメジャーレコード

マルチメジャーレコードを使用すると、時系列データをよりコンパクトな形式でメモリとマグネティックストアに保存できるため、データストレージコストを削減できます。また、コンパクトなデータストレージは、データ取得のためのより簡単なクエリの書き込みに役立ち、クエリのパフォーマンスを向上させ、クエリのコストを削減します。

さらに、マルチメジャーレコードは、時系列レコードに複数のタイムスタンプを保存するための TIMESTAMP データ型もサポートしています。マルチメジャーレコードの TIMESTAMP 属性は、将来または過去のタイムスタンプをサポートします。したがって、マルチメジャーレコードは、パフォーマンス、コスト、クエリのシンプルさの向上に役立ち、さまざまなタイプの相関メジャーをより柔軟に格納できます。

例: 動画ストリーミングアプリケーションのパフォーマンスと状態のモニタリング

200 個の EC2 インスタンスで実行されている動画ストリーミングアプリケーションを考えてみましょう。Amazon Timestream for LiveAnalytics を使用してアプリケーションから出力されるメトリクスを保存および分析すると、アプリケーションのパフォーマンスと状態を把握し、異常をすばやく特定して問題を解決し、最適化の機会を発見できます。

このシナリオを単一メジャーレコードとマルチメジャーレコードでモデリングし、両方のアプローチを比較/対照します。アプローチごとに、以下の前提があります。

単一メジャーレコードの使用

データモデリング: 単一メジャーレコードでは、4 つのメジャー (動画の起動時間、再バッファリング率、動画再生失敗、平均フレームレート) ごとに 1 つのレコードを作成します。各レコードには、4 つのディメンション (device_id、device_type、os_version、region) とタイムスタンプがあります。

書き込み: Amazon Timestream for LiveAnalytics にデータを書き込むと、レコードは次のように構築されます。

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 USD と推定されます。

マルチメジャーレコードの使用

データモデリング: マルチメジャーレコードでは、4 つのメジャーすべて (動画の起動時間、再バッファリング率、動画再生失敗、平均フレームレート)、4 つのディメンションすべて (device_id、device_type、os_version、region)、タイムスタンプを含む単一のレコードを作成します。

書き込み: Amazon Timestream for LiveAnalytics にデータを書き込むと、レコードは次のように構築されます。

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 USD と推定されます。

過去または将来に存在するタイムスタンプ付きのデータの書き込み

Timestream for LiveAnalytics では、いくつかの異なるメカニズムを使用して、メモリストアの保持期間外にあるタイムスタンプでデータを書き込むことができます。

読み取りの結果整合性

Timestream for LiveAnalytics は、読み取りの結果整合性セマンティクスをサポートします。つまり、データのバッチを Timestream for LiveAnalytics に書き込んだ直後にデータをクエリする場合、クエリ結果は最近完了した書き込みオペレーションの結果を反映していない可能性があります。少し時間が経ってからこれらのクエリリクエストを繰り返すと、結果で最新のデータが返されます。