開始方法: Amazon EventBridge パイプを作成する

パイプとその機能に慣れるために、 AWS CloudFormation テンプレートを使用して EventBridge パイプと関連するコンポーネントを設定します。その後、さまざまなパイプ機能を調べることができます。

テンプレートは、DynamoDB テーブルから Amazon SQS キューにストリームを接続する EventBridge パイプを作成します。データベーステーブルでレコードが作成または変更されるたびに、パイプは結果のイベントをキューに送信します。

デプロイされたパイプは、以下で構成されます。

パイプソースとして機能する DynamoDB テーブル (およびストリーム）、およびターゲットとしての Amazon SQS キュー。
DynamoDB テーブルと Amazon SQS キューにアクセスするために必要なアクセス許可を EventBridge に付与する実行ロール。
パイプ自体。テーブル項目の作成 (挿入) または変更時に生成されたイベントのみを選択するイベントフィルターが含まれています。

テンプレートの具体的な技術的詳細については、「」を参照してくださいテンプレートの詳細。

を使用したパイプの作成 AWS CloudFormation

パイプとその関連リソースを作成するには、CloudFormation テンプレートを作成し、それを使用してソースとターゲットを含むサンプルパイプを含むスタックを作成します。

重要

このテンプレートからスタックを作成する場合に使用される Amazon リソースに対して課金されます。

テンプレートの作成

まず、CloudFormation テンプレートを作成します。

テンプレートセクションで、JSON または YAML タブのコピーアイコンをクリックして、テンプレートの内容をコピーします。
テンプレートの内容を新しいファイルに貼り付けます。
ファイルをローカルに保存します。

スタックの作成

次に、保存したテンプレートを使用して CloudFormation スタックをプロビジョニングします。

AWS CloudFormation コンソールを開きます。
スタックページで、スタックの作成メニューから、新しいリソース (標準) で を選択します。
テンプレートを指定します。
1. 「前提条件」で、「既存のテンプレートを選択する」を選択します。
2. [テンプレートの指定] で、[テンプレートファイルのアップロード] を選択します。
3. ファイルの選択を選択し、テンプレートファイルに移動して選択します。
4. [次へ] を選択します。
スタックの詳細を指定します。
1. スタック名を入力します。
2. パラメータの場合は、デフォルト値を受け入れるか、独自の値を入力します。
3. [次へ] を選択します。
スタックオプションを設定します。
1. スタック障害オプションで、新しく作成されたリソースをすべて削除を選択します。
  
  注記
  このオプションを選択すると、スタックの作成が失敗した場合でも、削除ポリシーで保持が指定されているリソースに対して課金される可能性があります。詳細については、「 AWS CloudFormation ユーザーガイド」のDeletionPolicy「属性」を参照してください。
2. 他のすべてのデフォルト値を受け入れます。
3. 機能のチェックボックスをオンにして、CloudFormation がアカウントに IAM リソースを作成する可能性があることを確認します。
4. [次へ] を選択します。
スタックの詳細を確認し、送信を選択します。

AWS CloudFormation はスタックを作成します。スタックの作成が完了すると、スタックリソースを使用する準備が整います。スタックの詳細ページのリソースタブを使用して、がアカウントでプロビジョニングしたリソースを表示できます。

パイプ機能の探索

パイプが作成されたら、EventBridge コンソールを使用してパイプオペレーションとテストイベント配信を監視できます。

https://console.aws.amazon.com/events/home?#/pipes で EventBridge コンソールを開きます。
作成したパイプを選択します。

パイプの詳細ページでは、パイプコンポーネントセクションにパイプを構成するリソースが表示され、各コンポーネントの詳細を示すタブが表示されます。

パイプ用に作成した実行ロールは、「設定」タブの「アクセス許可」セクションにあります。

パイプフィルターの確認

パイプオペレーションをテストする前に、指定したフィルターを調べて、ターゲットに送信されるイベントを制御しましょう。パイプは、フィルター条件に一致するイベントのみをターゲットに送信します。それ以外のイベントはすべて破棄されます。この場合、テーブルエントリが作成または変更されたときにのみイベントが Amazon SQS キューに送信されるようにします。

パイプの詳細ページのパイプコンポーネントで、フィルタリングタブを選択します。

が INSERTまたは eventNameに設定されているイベントのみを選択するフィルターが含まれていますMODIFY。
```
{
  "eventName": ["INSERT", "MODIFY"]
}
```

パイプを介したイベントの送信

次に、パイプソースにイベントを生成して、パイプのフィルタリングと配信が正しく動作していることをテストします。これを行うには、パイプソースとして指定した DynamoDB テーブルで項目を作成および編集します。

パイプの詳細ページのパイプコンポーネントで、ソースタブを選択します。
Source で、DynamoDB ストリーム名を選択します。

これにより、DynamoDB コンソールが別のウィンドウで開き、ソーステーブルの詳細が表示されます。
[テーブルアイテムの探索] を選択します。
テーブルに項目を作成してINSERTイベントを生成します。
1. [項目を作成] を選択します。
2. アルバム属性とアーティスト属性の値を追加します。
3. [項目を作成] を選択します。
項目を編集して DELETEと INSERTイベントを生成します。
1. リストから項目を選択し、アクションメニューから項目の編集を選択します。
2. アルバムまたはアーティスト属性に新しい値を入力します。
3. 項目キーの値を変更することを確認するチェックボックスをオンにし、項目の再作成を選択します。
  
  これにより、項目が削除されて再作成され、DELETEイベントが生成され、新しいINSERTイベントが生成されます。
項目に属性を追加してMODIFYイベントを生成します。
1. リストから項目を選択し、アクションメニューから項目の編集を選択します。
2. 新しい属性の追加メニューから、数値を選択します。
3. 属性名に「年」と入力し、属性の値を入力します。[保存して閉じる] を選択します。

パイプを介したイベント配信の確認

最後に、パイプが DynamoDB でテーブル項目を作成および編集することで生成したイベントを正常にフィルタリングして配信したことを確認します。

パイプの詳細ページのパイプコンポーネントで、ターゲットタブを選択します。
Target で、Amazon SQS キュー名を選択します。

これにより、Amazon SQS コンソールが別のウィンドウで開き、ターゲットキューの詳細が表示されます。
[メッセージの送信と受信] を選択します。
「メッセージを受信する」で、「メッセージのポーリング」を選択します。

Amazon SQS は、受信したメッセージをキューにロードします。個々のメッセージをクリックすると、その詳細が表示されます。

キューには 3 つのイベントメッセージが必要です。
- タイプの 2 つINSERT。1 つはテーブル項目を最初に作成したときに生成され、もう 1 つはキー値を変更して項目を再作成したときに生成されます。
- タイプの 1 つMODIFY。テーブル項目に属性を追加したときに生成されます。
キー値を変更してテーブル項目を削除して再作成したときに生成されていても、キューDELETEにはタイプのイベントメッセージがないことに注意してください。指定したパイプフィルターは INSERTとでのみ選択されるためMODIFY、パイプはDELETEイベントをキューに配信するのではなく除外します。

クリーンアップ: リソースの削除

最後のステップとして、スタックとそれに含まれるリソースを削除します。

重要

スタックに含まれる Amazon リソースが存在する限り、課金されます。

AWS CloudFormation コンソールを開きます。
スタックページで、テンプレートから作成されたスタックを選択し、削除を選択し、削除を確認します。

CloudFormation は、スタックとそれに含まれるすべてのリソースの削除を開始します。

CloudFormation テンプレートの詳細

このテンプレートは、アカウントにリソースを作成し、アクセス許可を付与します。

リソース

このチュートリアルの AWS CloudFormation テンプレートは、アカウントに次のリソースを作成します。

重要

このテンプレートからスタックを作成する場合に使用される Amazon リソースに対して課金されます。

AWS::DynamoDB::Table: パイプのイベントソースとして機能する DynamoDB テーブル。
AWS::SQS::Queue: パイプを流れるイベントのターゲットとして機能する Amazon SQS キュー。
AWS::IAM::Role: アカウントの EventBridge Pipes サービスにアクセス許可を付与する IAM 実行ロール。
AWS::Pipes::Pipe: DynamoDB テーブルを Amazon SQS キューに接続するパイプ。

アクセス許可

テンプレートには、実行ロールを表す AWS::IAM::Roleリソースが含まれています。このロールは、EventBridge Pipes サービス (pipes.amazonaws.com) にアカウント内の次のアクセス許可を付与します。

次のアクセス許可は、DynamoDB テーブルにスコープされ、テンプレートがパイプのイベントソースとして作成するストリームです。

dynamodb:DescribeStream
dynamodb:GetRecords
dynamodb:GetShardIterator
dynamodb:ListStreams

次のアクセス許可は、スタックがパイプのターゲットとして作成する Amazon SQS キューに限定されます。

sqs:SendMessage

CloudFormation テンプレート

次の JSON または YAML コードを別のファイルとして保存し、このチュートリアルの CloudFormation テンプレートとして使用します。

JSON


{
  "AWSTemplateFormatVersion": "2010-09-09",

 "Description" : "EventBridge Pipe template example. Provisions a pipe, along with a DynamoDB stream as the pipe source and an SQS queue as the pipe target. Also provisions an execution role that contains the necessary permissions to access both the source and target. Once provisioned, the pipe receives events from the DynamoDB data stream, applies a filter, and sends matching events on to an SQS Queue. You will be billed for the Amazon resources used if you create a stack from this template.",

  "Parameters" : {
    "SourceTableName" : {
      "Type" : "String",
      "Default" : "pipe-example-source",
      "Description" : "Specify the name of the table to provision as the pipe source, or accept the default."
    },
  "TargetQueueName" : {
    "Type" : "String",
    "Default" : "pipe-example-target",
    "Description" : "Specify the name of the queue to provision as the pipe target, or accept the default."
  },
    "PipeName" : {
      "Type" : "String",
      "Default" : "pipe-with-filtering-example",
      "Description" : "Specify the name of the table to provision as the pipe source, or accept the default."
    }
},
  "Resources": {
    "PipeSourceDynamoDBTable": {
      "Type": "AWS::DynamoDB::Table",
      "Properties": {
        "AttributeDefinitions": [{
            "AttributeName": "Album",
            "AttributeType": "S"
          },
          {
            "AttributeName": "Artist",
            "AttributeType": "S"
          }

        ],
        "KeySchema": [{
            "AttributeName": "Album",
            "KeyType": "HASH"

          },
          {
            "AttributeName": "Artist",
            "KeyType": "RANGE"
          }
        ],
        "ProvisionedThroughput": {
          "ReadCapacityUnits": 10,
          "WriteCapacityUnits": 10
        },
        "StreamSpecification": {
          "StreamViewType": "NEW_AND_OLD_IMAGES"
        },
        "TableName": { "Ref" : "SourceTableName" }
      }
    },
    "PipeTargetQueue": {
      "Type": "AWS::SQS::Queue",
      "Properties": {
        "QueueName": { "Ref" : "TargetQueueName" }
      }
    },
    "PipeTutorialPipeRole": {
      "Type": "AWS::IAM::Role",
      "Properties": {
        "AssumeRolePolicyDocument": {
          "Version": "2012-10-17",
          "Statement": [{
            "Effect": "Allow",
            "Principal": {
              "Service": "pipes.amazonaws.com"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
              "StringLike": {
                "aws:SourceArn": {
	"Fn::Join": [
		"", 
                [
                  "arn:",
                  { "Ref": "AWS::Partition" },
                  ":pipes:",
                  { "Ref": "AWS::Region" },
                  ":",
                  { "Ref": "AWS::AccountId" },
                  ":pipe/",
                  { "Ref": "PipeName" }
               ]
	]
       },
                "aws:SourceAccount": { "Ref" : "AWS::AccountId" }
              }
            }
          }]
        },
        "Description" : "EventBridge Pipe template example. Execution role that grants the pipe the permissions necessary to send events to the specified pipe.",
        "Path": "/",
        "Policies": [{
            "PolicyName": "SourcePermissions",
            "PolicyDocument": {
              "Version": "2012-10-17",
              "Statement": [{
                "Effect": "Allow",
                "Action": [
                  "dynamodb:DescribeStream",
                  "dynamodb:GetRecords",
                  "dynamodb:GetShardIterator",
                  "dynamodb:ListStreams"
                ],
                "Resource": [
                  { "Fn::GetAtt" : [ "PipeSourceDynamoDBTable", "StreamArn" ] }
                ]
              }]
            }
          },
          {
            "PolicyName": "TargetPermissions",
            "PolicyDocument": {
              "Version": "2012-10-17",
              "Statement": [{
                "Effect": "Allow",
                "Action": [
                  "sqs:SendMessage"
                ],
                "Resource": [
                  { "Fn::GetAtt" : [ "PipeTargetQueue", "Arn" ] }
                ]
              }]
            }
          }
        ]
      }
  },
    "PipeWithFiltering": {
      "Type": "AWS::Pipes::Pipe",
      "Properties": {
        "Description" : "EventBridge Pipe template example. Pipe that receives events from a DynamoDB stream, applies a filter, and sends matching events on to an SQS Queue.",
        "Name": { "Ref" : "PipeName" },
        "RoleArn": {"Fn::GetAtt" : ["PipeTutorialPipeRole", "Arn"] },
        "Source": { "Fn::GetAtt" : [ "PipeSourceDynamoDBTable", "StreamArn" ] },
        "SourceParameters": {
          "DynamoDBStreamParameters" : {
            "StartingPosition" : "LATEST"
         },
        "FilterCriteria" : {
          "Filters" : [ {
            "Pattern" : "{ \"eventName\": [\"INSERT\", \"MODIFY\"] }"
         }]
        }
        },
        "Target": { "Fn::GetAtt" : [ "PipeTargetQueue", "Arn" ] }
      }
    }
  }
}

YAML


AWSTemplateFormatVersion: '2010-09-09'
Description: EventBridge Pipe template example. Provisions a pipe, along with a DynamoDB stream as the pipe source and an SQS queue as the pipe target. Also provisions an execution role that contains the necessary permissions to access both the source and target. Once provisioned, the pipe receives events from the DynamoDB data stream, applies a filter, and sends matching events on to an SQS Queue. You will be billed for the Amazon resources used if you create a stack from this template.
Parameters:
  SourceTableName:
    Type: String
    Default: pipe-example-source
    Description: Specify the name of the table to provision as the pipe source, or accept the default.
  TargetQueueName:
    Type: String
    Default: pipe-example-target
    Description: Specify the name of the queue to provision as the pipe target, or accept the default.
  PipeName:
    Type: String
    Default: pipe-with-filtering-example
    Description: Specify the name of the table to provision as the pipe source, or accept the default.
Resources:
  PipeSourceDynamoDBTable:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
        - AttributeName: Album
          AttributeType: S
        - AttributeName: Artist
          AttributeType: S
      KeySchema:
        - AttributeName: Album
          KeyType: HASH
        - AttributeName: Artist
          KeyType: RANGE
      ProvisionedThroughput:
        ReadCapacityUnits: 10
        WriteCapacityUnits: 10
      StreamSpecification:
        StreamViewType: NEW_AND_OLD_IMAGES
      TableName: !Ref SourceTableName
  PipeTargetQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Ref TargetQueueName
  PipeTutorialPipeRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: pipes.amazonaws.com
            Action: sts:AssumeRole
            Condition:
              StringLike:
                aws:SourceArn: !Join
                  - ''
                  - - 'arn:'
                    - !Ref AWS::Partition
                    - ':pipes:'
                    - !Ref AWS::Region
                    - ':'
                    - !Ref AWS::AccountId
                    - ':pipe/'
                    - !Ref PipeName
                aws:SourceAccount: !Ref AWS::AccountId
      Description: EventBridge Pipe template example. Execution role that grants the pipe the permissions necessary to send events to the specified pipe.
      Path: /
      Policies:
        - PolicyName: SourcePermissions
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - dynamodb:DescribeStream
                  - dynamodb:GetRecords
                  - dynamodb:GetShardIterator
                  - dynamodb:ListStreams
                Resource:
                  - !GetAtt PipeSourceDynamoDBTable.StreamArn
        - PolicyName: TargetPermissions
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - sqs:SendMessage
                Resource:
                  - !GetAtt PipeTargetQueue.Arn
  PipeWithFiltering:
    Type: AWS::Pipes::Pipe
    Properties:
      Description: EventBridge Pipe template example. Pipe that receives events from a DynamoDB stream, applies a filter, and sends matching events on to an SQS Queue.
      Name: !Ref PipeName
      RoleArn: !GetAtt PipeTutorialPipeRole.Arn
      Source: !GetAtt PipeSourceDynamoDBTable.StreamArn
      SourceParameters:
        DynamoDBStreamParameters:
          StartingPosition: LATEST
        FilterCriteria:
          Filters:
            - Pattern: '{ "eventName": ["INSERT", "MODIFY"] }'
      Target: !GetAtt PipeTargetQueue.Arn

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

入門

イベントバス