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

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

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

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

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

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

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

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

重要

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

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

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

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

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

CloudFormation を使用してスタックを作成する (AWS CLI）

を使用してスタック AWS CLI を作成することもできます。

create-stack コマンドを使用します。

スタック名を指定して、デフォルトのテンプレートパラメータ値を受け入れます。template-body パラメータを使用してテンプレートコンテンツを渡すか、URL の場所template-urlを指定します。
```
aws cloudformation create-stack \
  --stack-name eventbridge-rule-tutorial \
  --template-body template-contents \
  --capabilities CAPABILITY_IAM
```

1 つ以上のテンプレートパラメータのデフォルト値 (複数可) を上書きします。例:


aws cloudformation create-stack \
  --stack-name eventbridge-rule-tutorial \
  --template-body template-contents \
  --parameters \
    ParameterKey=SourceTableName,ParameterValue=pipe-example-source \
    ParameterKey=TargetQueueName,ParameterValue=pipe-example-target \
    ParameterKey=PipeName,ParameterValue=pipe-with-filtering-example \
  --capabilities CAPABILITY_IAM

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 リソースが存在する限り、課金されます。

クラウドフォーメーションコンソールのhttps://console.aws.amazon.com/cloudformation/ 開きます：
スタックページで、テンプレートから作成されたスタックを選択し、削除を選択し、削除を確認します。

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

CloudFormation テンプレートの詳細

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

リソース

このチュートリアルの 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" : "[AWSDocs] EventBridge: pipes-get-started",

  "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: '[AWSDocs] EventBridge: pipes-get-started'
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 を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

イベントバスの使用を開始する

イベントバス