Kinesis での QLDB ストリーミングレコード

Amazon QLDB ストリーミングは、コントロール、ブロックサマリー、およびリビジョンの詳細という 3 種類のデータレコードを指定の Amazon Kinesis Data Streams リソースに書き込みます。3 つのレコードタイプはすべて、Amazon Ion 形式のバイナリ表現で記述されます。

コントロールレコードは、QLDB ストリーミングの開始と完了を示します。QLDB ストリーミングは、リビジョンがジャーナルにコミットされるたびに、関連するすべてのジャーナルブロックデータをブロックサマリーおよびリビジョンの詳細レコードに書き込みます。

3 つのレコードタイプは多相型です。これらはすべて、QLDB ストリーミング ARN、レコードタイプ、およびレコードペイロードを含む共通の最上位レコードで構成されます。この最上位レベルのレコードは、次の形式になります。

{
  qldbStreamArn: string,
  recordType: string,
  payload: {
    //control | block summary | revision details record
  }
}

recordType フィールドには、次の 3 つの値のいずれかを指定できます。

• CONTROL

• BLOCK_SUMMARY

• REVISION_DETAILS

次のセクションでは、個々のペイロードレコードの形式と内容について説明します。

注記

QLDB は、Amazon Ion のバイナリ表現で Kinesis Data Streams にすべてのストリーミングレコードを書き込みます。以下の例は、Ion のテキスト表現で、レコードの内容を読みやすい形式で説明しています。

コントロールレコード

QLDB ストリーミングは、開始イベントと完了イベントを示すコントロールレコードを書き込みます。次に、それぞれの controlRecordType のサンプルデータを含むコントロールレコードの例を示します。

• CREATED - 新しく作成されたストリーミングがアクティブであることを示すために、QLDB ストリーミングが Kinesis に書き込む最初のレコード。

  {
    qldbStreamArn:"arn:aws:qldb:us-east-1:123456789012:stream/exampleLedger/IiPT4brpZCqCq3f4MTHbYy",
    recordType:"CONTROL",
    payload:{
      controlRecordType:"CREATED"
    }
  }

• COMPLETED - ストリーミングが指定された終了日時に達したこと示すために、QLDB ストリーミングが Kinesis に書き込む最後のレコード。ストリームをキャンセルすると、このレコードは書き込まれません。

  {
    qldbStreamArn:"arn:aws:qldb:us-east-1:123456789012:stream/exampleLedger/IiPT4brpZCqCq3f4MTHbYy",
    recordType:"CONTROL",
    payload:{
      controlRecordType:"COMPLETED"
    }
  }

ブロックサマリーレコード

ブロックサマリーレコードは、ドキュメントのリビジョンがコミットされるジャーナルブロックを表します。ブロックは、トランザクション中に QLDB ジャーナルにコミットされるオブジェクトです。

ブロックサマリーレコードのペイロードには、ブロックをコミットしたトランザクションのブロックアドレス、タイムスタンプ、およびその他のメタデータが含まれます。また、ブロック内のリビジョンのサマリー属性と、それらをコミットした PartiQL ステートメントも含まれます。以下は、サンプルデータを含むブロックサマリーレコードの例です。

注記

このブロックサマリーの例は、情報提供のみを目的として記載されています。示されているハッシュは、実際に計算されたハッシュ値ではありません。

{
  qldbStreamArn:"arn:aws:qldb:us-east-1:123456789012:stream/exampleLedger/IiPT4brpZCqCq3f4MTHbYy",
  recordType:"BLOCK_SUMMARY",
  payload:{
    blockAddress:{
      strandId:"ElYL30RGoqrFCbbaQn3K6m",
      sequenceNo:60807
    },
    transactionId:"9RWohCo7My4GGkxRETAJ6M",
    blockTimestamp:2019-09-18T17:00:14.601000001Z,
    blockHash:{{6Pk9KDYJd38ci09oaHxx0D2grtgh4QBBqbDS6i9quX8=}},
    entriesHash:{{r5YoH6+NXDXxgoRzPREGAWJfn73KlZE0eTfbTxZWUDU=}},
    previousBlockHash:{{K3ti0Agk7DEponywKcQCPRYVHb5RuyxdmQFTfrloptA=}},
    entriesHashList:[
      {{pbzvz6ofJC7mD2jvgfyrY/VtR01zIZHoWy8T1VcxlGo=}},
      {{k2brC23DLMercmiOWHiURaGwHu0mQtLzdNPuviE2rcs=}},
      {{hvw1EV8k4oOkIO36kblO/+UUSFUQqCanKuDGr0aP9nQ=}},
      {{ZrLbkyzDcpJ9KWsZMZqRuKUkG/czLIJ4US+K5E31b+Q=}}
    ],
    transactionInfo:{
      statements:[
        {
          statement:"SELECT * FROM Person WHERE GovId = ?",
          startTime:2019-09-18T17:00:14.587Z,
          statementDigest:{{p4Dn0DiuYD3Xm9UQQ75YLwmoMbSfJmopOmTfMnXs26M=}}
        },
        {
          statement:"INSERT INTO Person ?",
          startTime:2019-09-18T17:00:14.594Z,
          statementDigest:{{klMLkLfa5VJqk6JUPtHkQpOsDdG4HmuUaq/VaApQflU=}}
        },
        {
          statement:"INSERT INTO VehicleRegistration ?",
          startTime:2019-09-18T17:00:14.598Z,
          statementDigest:{{B0gO9BWVNrzRYFoe7t+GVLpJ6uZcLKf5t/chkfRhspI=}}
        }
      ],
      documents:{
        '7z2OpEBgVCvCtwvx4a2JGn':{
          tableName:"Person",
          tableId:"LSkFkQvkIOjCmpTZpkfpn9",
          statements:[1]
        },
        'K0FpsSLpydLDr7hi6KUzqk':{
          tableName:"VehicleRegistration",
          tableId:"Ad3A07z0ZffC7Gpso7BXyO",
          statements:[2]
        }
      }
    },
    revisionSummaries:[
      {
        hash:{{uDthuiqSy4FwjZssyCiyFd90XoPSlIwomHBdF/OrmkE=}},
        documentId:"7z2OpEBgVCvCtwvx4a2JGn"
      },
      {
        hash:{{qJID/amu0gN3dpG5Tg0FfIFTh/U5yFkfT+g/O6k5sPM=}},
        documentId:"K0FpsSLpydLDr7hi6KUzqk"
      }
    ]
  }
}

revisionSummaries フィールドでは、一部のリビジョンに documentId がない場合があります。これらは、ユーザーデータを含まない内部のみのシステムリビジョンです。これらのリビジョンのハッシュはジャーナルのハッシュチェーン全体の一部であるため、QLDB ストリームでは、リビジョンが対応するブロックサマリーレコードに含まれます。暗号検証には、完全なハッシュチェーンが必要です。

次のセクションで説明するように、ドキュメント ID を持つリビジョンだけが、個別のリビジョン詳細レコードで発行されます。

リビジョン詳細レコード

リビジョン詳細レコードは、ジャーナルにコミットされたドキュメントのリビジョンを表します。ペイロードには、リビジョンのコミットされたビューのすべての属性と、関連付けられたテーブル名とテーブル ID が含まれます。次に、サンプルデータを含むリビジョンレコードの例を示します。

{
  qldbStreamArn:"arn:aws:qldb:us-east-1:123456789012:stream/exampleLedger/IiPT4brpZCqCq3f4MTHbYy",
  recordType:"REVISION_DETAILS",
  payload:{
    tableInfo:{
      tableName:"VehicleRegistration",
      tableId:"Ad3A07z0ZffC7Gpso7BXyO"
    },
    revision:{
      blockAddress:{
        strandId:"ElYL30RGoqrFCbbaQn3K6m",
        sequenceNo:60807
      },
      hash:{{qJID/amu0gN3dpG5Tg0FfIFTh/U5yFkfT+g/O6k5sPM=}},
      data:{
        VIN:"1N4AL11D75C109151",
        LicensePlateNumber:"LEWISR261LL",
        State:"WA",
        City:"Seattle",
        PendingPenaltyTicketAmount:90.25,
        ValidFromDate:2017-08-21,
        ValidToDate:2020-05-11,
        Owners:{
          PrimaryOwner:{PersonId:"7z2OpEBgVCvCtwvx4a2JGn"},
          SecondaryOwners:[]
        }
      },
      metadata:{
        id:"K0FpsSLpydLDr7hi6KUzqk",
        version:0,
        txTime:2019-09-18T17:00:14.602Z,
        txId:"9RWohCo7My4GGkxRETAJ6M"
      }
    }
  }
}

重複レコードと out-of-order レコードの処理

QLDB ストリームは、重複する レコードと out-of-order レコードを Kinesis Data Streams に発行できます。したがって、コンシューマーアプリケーションは、そのようなシナリオを特定して処理するために、独自のロジックを実装する必要があります。ブロックサマリーレコードとリビジョン詳細レコードには、この目的で使用できるフィールドが含まれます。これらのフィールドは、ダウンストリームサービスの機能と組み合わせて、一意の ID とレコードの厳密な順序の両方を示すことができます。

例えば、QLDB を OpenSearch インデックスと統合して、ドキュメントに対してフルテキスト検索機能を提供するストリームがあるとします。このユースケースでは、ドキュメントの古い (out-of-order) リビジョンのインデックスを作成しないようにする必要があります。順序付けと重複排除を適用するには、 OpenSearch のドキュメント ID および外部バージョンフィールドと、リビジョンの詳細レコードのドキュメント ID およびバージョンフィールドを使用します。

QLDB を Amazon OpenSearch Service と統合するサンプルアプリケーションの重複排除ロジックの例については、 GitHub リポジトリ aws-samples/amazon-qldb-streaming-amazon- opensearch-service-sample-pythonを参照してください。