Amazon EventBridge Pipes のバッチ処理と同時実行

バッチ処理動作

EventBridge Pipes は、ソースから、そしてそれをサポートするターゲットへのバッチ処理に対応しています。また、AWS Lambda および AWS Step Functions では、エンリッチメントのバッチ処理もサポートされています。サービスが異なればサポートされるバッチ処理のレベルも異なるため、ターゲットがサポートするバッチサイズよりも大きいバッチサイズのパイプを構成することはできません。例えば、Amazon Kinesis ストリームソースは、最大バッチサイズ 10,000 件のレコードをサポートしますが、Amazon Simple Queue Service は、バッチあたり最大 10 件のメッセージをターゲットとしてサポートします。そのため、Kinesis ストリームから Amazon SQS キューへのパイプは、ソースの最大バッチサイズを 10 に設定できます。

バッチ処理をサポートしないエンリッチメントまたはターゲットを使用してパイプを設定すると、ソースでバッチ処理を有効にすることはできません。

ソースでバッチ処理が有効化されると、JSON レコードの配列がパイプを通過し、サポートされているエンリッチメントまたはターゲットのバッチ API にマッピングされます。Input transformers は、配列全体ではなく、配列内の個々の JSON レコードに個別に適用されます。これらの配列の例については、Amazon EventBridge Pipes ソース を参照し、特定のソースを選択してください。パイプは、バッチサイズが 1 の場合でも、サポートされているエンリッチメントまたはターゲットにバッチ API を使用します。エンリッチメントまたはターゲットにバッチ API がないものの、Lambda や Step Functions などの完全な JSON ペイロードを受け取る場合、JSON 配列全体が 1 回のリクエストで送信されます。バッチサイズが 1 の場合でも、リクエストは JSON 配列として送信されます。

パイプがソースでバッチ処理するように設定されていて、ターゲットがバッチ処理をサポートしている場合、エンリッチメントから JSON 項目の配列を返すことができます。この配列は、元のソースよりも短い配列でも長い配列でもかまいません。ただし、配列がターゲットがサポートするバッチサイズよりも大きい場合、パイプはターゲットを呼び出しません。

サポートされているバッチ処理可能なターゲット

次のエンリッチメントとターゲットは、バッチイベントペイロード全体を受け取って処理しますが、バッチのサイズではなく、イベントの合計ペイロードサイズによって制約されます。

部分的なバッチ処理失敗

Amazon SQS および Kinesis や DynamoDB などのストリームソースの場合、EventBridge Pipes はターゲット障害の部分的なバッチ障害処理をサポートします。ターゲットがバッチ処理をサポートしていて、バッチの一部のみが成功した場合、EventBridge はペイロードの残りの部分のバッチ処理を自動的に再試行します。最新のエンリッチメントコンテンツの場合、この再試行は、設定されたエンリッチメントの再呼び出しを含め、パイプ全体で行われます。

エンリッチメントの部分的なバッチ障害処理はサポートされていません。

Lambda と Step Functions ターゲットの場合、ターゲットから構造を定義したペイロードを返すことで部分的な障害を指定することもできます。これは再試行が必要なイベントを示しています。

部分的障害ペイロード構造の例

{ 
  "batchItemFailures": [ 
    {
      "itemIdentifier": "id2"
    },
    {
      "itemIdentifier": "id4"
    }
]

この例では、itemIdentifier が、元のソースからのターゲットにより処理されるイベントの ID と一致しています。Amazon SQS の場合、これは messageId です。Kinesis と DynamoDB の場合、これは eventID です。EventBridge Pipes がターゲットからの部分的なバッチ障害を適切に処理するには、これらのフィールドをエンリッチメントによって返される配列ペイロードに含める必要があります。

スループットと同時実行動作

パイプが受信したエンリッチメントまたはターゲットに送信されるすべてのイベントまたはイベントのバッチは、パイプの実行と見なされます。STARTED 状態のパイプは、ソースからのイベントを継続的にポーリングし、利用可能なバックログと設定されたバッチ設定に応じてスケールアップとスケールダウンを行います。

パイプの同時実行のクォータ、およびアカウントとリージョンごとのパイプ数については、EventBridge Pipes クォータ を参照してください。

デフォルトでは、1 つのパイプは、ソースに応じて次の最大同時実行数にスケーリングされます。

最大ポーリングスループットまたは同時実行数の制限をより高くする必要がある場合は、サポートにお問い合わせください。

パイプの実行は、エンリッチメント処理とターゲット処理を含めて最大 5 分に制限されています。この制限を引き上げることはできません。

ソースが厳密に順序付けされたパイプ (Amazon SQS FIFO キュー、Kinesis および DynamoDB Streams、Apache Kafka トピックなど) は、FIFO キューのメッセージグループ ID 数や Kinesis キューのシャード数など、ソースの設定によって同時実行性はさらに制限されます。順序付けはこれらの制約の範囲内で厳密に保証されているため、順序付けされたソースを含むパイプは、これらの同時実行制限を超えることはできません。