翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

writes

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

データは、AWS SDKs、、または AWS CLI、、、AWS LambdaAWS IoT CoreAmazon Managed Service for Apache FlinkAmazon KinesisAmazon MSK、 を使用して書き込むことができますオープンソース Telegraf。

データ型

Timestream for Live Analytics では、書き込みに次のデータ型がサポートされています。

事前スキーマ定義なし

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

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

Amazon Timestream for Live Analytics の書き込みオペレーションを使用すると、データを挿入してアップサートできます。デフォルトでは、Amazon Timestream for Live Analytics での書き込みは、最初のライターがセマンティクスを獲得します。セマンティクスでは、データは追加としてのみ保存され、重複レコードは拒否されます。最初のライターがセマンティクスを勝ち取り、多くの時系列アプリケーションの要件を満たしていますが、アプリケーションが冪等な方法で既存のレコードを更新したり、最後のライターがセマンティクスを勝ち取ったデータを書き込んだりする必要があり、バージョンが最も高いレコードがサービスに保存されるシナリオがあります。これらのシナリオに対応するため、Amazon Timestream for Live Analytics ではデータをアップサートできます。Upsert は、レコードが存在しない場合はシステムにレコードを挿入し、存在する場合はレコードを更新するオペレーションです。レコードが更新されると、べき等な方法で更新されます。

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

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

Amazon Timestream for Live Analytics では、メモリストアとマグネティックストアにデータを直接書き込む機能を提供しています。メモリストアは高スループットのデータ書き込み用に最適化されており、マグネティックストアは遅延到着データのスループット書き込みを減らすように最適化されています。

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

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

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

単一メジャーレコード

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

複数メジャーレコード

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

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

複数メジャーレコード

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

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

例: ビデオストリーミングアプリケーションのパフォーマンスとヘルスのモニタリング

200 EC2 インスタンスで実行されているビデオストリーミングアプリケーションについて考えてみましょう。Amazon Timestream for Live Analytics を使用して、アプリケーションから出力されるメトリクスを保存および分析し、アプリケーションのパフォーマンスと正常性を理解し、異常を迅速に特定し、問題を解決し、最適化の機会を発見できるようにします。

このシナリオを単一メジャーレコードと複数メジャーレコードでモデル化し、両方のアプローチを比較/コントラストします。各アプローチについて、以下の仮定を行います。

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

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

書き込み: ライブ分析用に 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

ワークロードコスト: このワークロードのコストは、単一メジャーレコードでは 1 か月あたり 373.23 USD と推定されます。

複数メジャーレコードの使用

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

書き込み: ライブ分析用に 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 USD と推定されます。

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

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

読み取りの結果整合性

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