Amazon EventBridge パイプの作成 - Amazon EventBridge
ソースの指定フィルタリングの設定エンリッチメントの定義ターゲットの設定パイプ設定の指定構成パラメータの検証

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

Amazon EventBridge パイプの作成

EventBridge Pipes では、高度なイベント変換とエンリッチメントを含む、ソースとターゲット間のポイントツーポイント統合を作成できます。

EventBridge パイプを作成するには、次のステップを実行します。

  1. ソースの指定

  2. イベントフィルタリングの設定 (オプション)

  3. イベントのエンリッチメントの定義 (オプション)

  4. ターゲットの設定

  5. パイプ設定の指定

サンプルパイプを迅速にセットアップするには、「開始方法: Amazon EventBridge パイプを作成する」を参照してください。このトピックでは、CloudFormation を使用してパイプとその関連リソースをデプロイし、パイプの機能の概要を説明します。

AWS CLI を使用してパイプを作成する方法については、「AWS CLI コマンドリファレンス」の「create-pipe」を参照してください。

ソースの指定

まず、パイプでイベントを受信するソースを指定します。

コンソールを使用してパイプソースを指定するには

  1. Amazon EventBridge コンソールの https://console.aws.amazon.com/events/ を開いてください。

  2. ナビゲーションペインで、[パイプ] を選択します。

  3. [Create pipe] (パイプの作成) を選択します。

  4. パイプの名前を入力します。

  5. (オプション) パイプの説明を追加します。

  6. [パイプを構築] タブの [ソース] で、このパイプに指定するソースのタイプを選択し、ソースを設定します。

    設定のプロパティは、選択するソースの種類によって異なります。

    Confluent
    コンソールを使用して Confluent Cloud ストリームをソースとして設定するには

    1. [ソース] で、[Confluent Cloud] を選択します。

    2. [Bootstrap servers] (ブートストラップサーバー) には、ブローカーの host:port ペアアドレスを入力します。

    3. [Topic name] (トピック名) には、読み取り元のトピック名を入力します。

    4. (オプション) [VPC] では、使用する VPC を選択します。次に、[VPC subnets] (VPC サブネット) に、目的のサブネットを選択します。[VPCSecurity groups] (VPC セキュリティグループ) で、セキュリティグループを選択します。

    5. [認証 - オプション] で、[認証を使用] をオンにして次の操作を行います。

      1. [Authentication method] (認証方法) で、認証タイプを選択します。

      2. [Secret key] (シークレットキー) で、シークレットキーを選択します。

      詳細については、「Confluent Documentation」の「Authenticate to Confluent Cloud resources」を参照してください。

    6. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Starting position] (開始位置) で、次のいずれかを選択します。

        • Latest (最新) — シャード内の最新のレコードでストリームの読み取りを開始します。

        • Trim horizon (水平トリム) — シャード内の最後のトリミングされていないレコードからストリームの読み取りを開始します。これはシャード内の最も古いレコードです。

      2. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大レコード数を入力します。デフォルト値は 100 です。

      3. [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

    DynamoDB

    1. [ソース] で、[DynamoDB] を選択します。

    2. DynamoDB ストリームの場合は、ソースとして使用するストリームを選択します。

    3. [Starting position] (開始位置) で、次のいずれかを選択します。

      • Latest (最新) — シャード内の最新のレコードでストリームの読み取りを開始します。

      • Trim horizon (水平トリム) — シャード内の最後のトリミングされていないレコードからストリームの読み取りを開始します。これはシャード内の最も古いレコードです。

    4. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大レコード数を入力します。デフォルト値は 100 です。

      2. [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

      3. [Concurrent batches per shard - optional] (シャードあたりの同時バッチ数 - オプション) の場合、同じシャードから同時に読み取ることができるバッチの数を入力します。

      4. [On partial batch item failure] (バッチ項目の一部に障害が発生した場合) では、次の項目を選択します。

        • AUTOMATIC_BISECT — すべてのレコードが処理されるか、バッチに 1 つの失敗メッセージが残るまで、各バッチを半分ずつ再試行します。

        注記

        AUTOMATIC_BISECT を選択しない場合は、特定の失敗したレコードを返すことができ、そのレコードだけが再試行されます。

    Kinesis
    コンソールを使用して Kinesis ソースを設定するには

    1. [ソース] で、[Kinesis] を選択します。

    2. [Kinesis stream] (Kinesis ストリーム) には、ソースとして使用するストリームを選択します。

    3. [Starting position] (開始位置) で、次のいずれかを選択します。

      • Latest (最新) — シャード内の最新のレコードでストリームの読み取りを開始します。

      • Trim horizon (水平トリム) — シャード内の最後のトリミングされていないレコードからストリームの読み取りを開始します。これはシャード内の最も古いレコードです。

      • [At timestamp] (タイムスタンプ時) — 指定した時刻からストリームの読み取りを開始します。[Timestamp] (タイムスタンプ) に、YYYY/MM/DD と hh:mm:ss 形式でデータと時刻を入力します。

    4. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大レコード数を入力します。デフォルト値は 100 です。

      2. (オプション) [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

      3. [Concurrent batches per shard - optional] (シャードあたりの同時バッチ数 - オプション) の場合、同じシャードから同時に読み取ることができるバッチの数を入力します。

      4. [On partial batch item failure] (バッチ項目の一部に障害が発生した場合) では、次の項目を選択します。

        • AUTOMATIC_BISECT — すべてのレコードが処理されるか、バッチに 1 つの失敗メッセージが残るまで、各バッチを半分ずつ再試行します。

        注記

        AUTOMATIC_BISECT を選択しない場合は、特定の失敗したレコードを返すことができ、そのレコードだけが再試行されます。

    Amazon MQ
    コンソールを使用して Amazon MQ ソースを設定するには

    1. [ソース] で、[Amazon MQ] を選択します。

    2. Amazon MQ ブローカーの場合は、ソースとして使用するストリームを選択します。

    3. [Queue name] (キュー名) には、読み取り元のトピック名を入力します。

    4. [Authentication Method] (認証方法) には、[BASIC_AUTH] を選択します。

    5. [Secret key] (シークレットキー) で、シークレットキーを選択します。

    6. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大メッセージ数を入力します。デフォルト値は 100 です。

      2. [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

    Amazon MSK
    コンソールを使用して Amazon MSK ソースを設定するには

    1. [ソース] で、[Amazon MSK] を選択します。

    2. [Amazon MSK cluster] (Amazon MSK クラスター) で、開くクラスターを選択します。

    3. [Topic name] (トピック名) には、読み取り元のトピック名を入力します。

    4. (オプション) [Consumer Group ID - optional] (コンシューマーグループ ID) で、パイプが参加するコンシューマーグループの ID を入力します。

    5. (オプション) [Authentication - optional] (認証 - オプション) で、[Use Authentication] (認証を使用) をオンにして次の操作を行います。

      1. [Authentication method] (認証方法) で、認証タイプを選択します。

      2. [Secret key] (シークレットキー) で、シークレットキーを選択します。

    6. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大レコード数を入力します。デフォルト値は 100 です。

      2. [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

      3. [Starting position] (開始位置) で、次のいずれかを選択します。

        • Latest (最新) — シャード内の最新のレコードでトピックの読み取りを開始します。

        • Trim horizon (水平トリム) — シャード内の最後のトリミングされていないレコードからトピックの読み取りを開始します。これはシャード内の最も古いレコードです。

          注記

          Trim horizon (水平トリム) はApache Kafka の[Earliest] (最も早い時間) と同じです。

    Self managed Apache Kafka
    コンソールを使用してセルフマネージド型の Apache Kafka ソースを設定するには

    1. [ソース] で、[セルフマネージド型の Apache Kafka] を選択します。

    2. [Bootstrap servers] (ブートストラップサーバー) には、ブローカーの host:port ペアアドレスを入力します。

    3. [Topic name] (トピック名) には、読み取り元のトピック名を入力します。

    4. (オプション) [VPC] では、使用する VPC を選択します。次に、[VPC subnets] (VPC サブネット) に、目的のサブネットを選択します。[VPCSecurity groups] (VPC セキュリティグループ) で、セキュリティグループを選択します。

    5. (オプション) [Authentication - optional] (認証 - オプション) で、[Use Authentication] (認証を使用) をオンにして次の操作を行います。

      1. [Authentication method] (認証方法) で、認証タイプを選択します。

      2. [Secret key] (シークレットキー) で、シークレットキーを選択します。

    6. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Starting position] (開始位置) で、次のいずれかを選択します。

        • Latest (最新) — シャード内の最新のレコードでストリームの読み取りを開始します。

        • Trim horizon (水平トリム) — シャード内の最後のトリミングされていないレコードからストリームの読み取りを開始します。これはシャード内の最も古いレコードです。

      2. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大レコード数を入力します。デフォルト値は 100 です。

      3. [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

    Amazon SQS
    コンソールを使用して Amazon SQS ソースを設定するには

    1. [ソース] で [SQS] を選択します。

    2. [SQS Queue] (SQS キュー) には、使用するキューを選択します。

    3. (オプション) [Additional settings - optional] (追加設定 - オプション) では、以下を実行します。

      1. [Batch size - optional] (バッチサイズ - オプション) に、各バッチの最大レコード数を入力します。デフォルト値は 100 です。

      2. [Batch window - optional] (バッチウィンドウ - オプション) の場合、次に進む前にレコードを収集する最大秒数を入力します。

イベントフィルタリングの設定 (オプション)

パイプにフィルタリングを追加して、ソースからイベントのサブセットのみをターゲットに送信できます。

コンソールを使用してフィルタリングを設定するには

  1. [Filtering] (フィルタリング) を選択します。

  2. [サンプルイベント - オプション] には、イベントパターンの作成に使用できるサンプルイベントが表示されます。また、[独自のイベントを入力] を選択して独自のイベントを入力することもできます。

  3. [イベントパターン] に、イベントのフィルタリングに使用するイベントパターンを入力します。フィルターの作成の詳細については、「Amazon EventBridge Pipes フィルタリング」を参照してください。

    以下は、[City] (都市) フィールドの値が [Seattle] のイベントのみを送信するイベントパターンの例です。

    {
  "data": {
    "City": ["Seattle"]
  }
}

イベントがフィルタリングされたので、オプションのエンリッチメントとパイプのターゲットを追加できます。

イベントのエンリッチメントの定義 (オプション)

エンリッチメント用のイベントデータを Lambda 関数、AWS Step Functions ステートマシン、Amazon API Gateway、または API 送信先に送信できます。

エンリッチメントを選択するには

  1. [Enrichment] (エンリッチメント) を選択します。

  2. [Details] (詳細) の [Service] (サービス) で、エンリッチメントに使用したいサービスと関連設定を選択します。

データを送信する前にデータを変換して拡張することもできます。

(オプション) Input Transformer を定義するには

  1. [Enrichment Input Transformer - optional] (エンリッチメント Input Transformer - オプション) を選択してください。

  2. [Sample events/Event Payload] (サンプルイベント/イベントペイロード)では、サンプルイベントタイプを選択します。

  3. [Transformer] (トランスフォーマー) には、サンプルイベントからフィールドへ参照する"Event happened at <$.detail.field>."場所<$.detail.field> など、トランスフォーマー構文を入力します。また、サンプルイベントのフィールドをダブルクリックしてトランスフォーマーに追加することもできます。

  4. [Output] (出力) で、出力が希望どおりであることを確認します。

これでデータのフィルタリングと拡張が完了したので、次はイベントデータを送信するターゲットを定義する必要があります。

ターゲットの設定

ターゲットを設定するには

  1. [Target] を選択します。

  2. [Details] (詳細) の [Target service] (ターゲットサービス) で、ターゲットを選択します。表示されるフィールドは、選択したターゲットによって異なります。必要に応じて、このターゲットタイプに固有の情報を入力します。

ターゲットに送信する前にデータを変換することもできます。

(オプション) Input Transformer を定義するには

  1. [Target Input Transformer - optional] (ターゲットインプットトランスフォーマー - オプション) を選択します。

  2. [Sample events/Event Payload] (サンプルイベント/イベントペイロード)では、サンプルイベントタイプを選択します。

  3. [Transformer] (トランスフォーマー) には、サンプルイベントからフィールドへ参照する"Event happened at <$.detail.field>."場所<$.detail.field> など、トランスフォーマー構文を入力します。また、サンプルイベントのフィールドをダブルクリックしてトランスフォーマーに追加することもできます。

  4. [Output] (出力) で、出力が希望どおりであることを確認します。

これでパイプが設定されたので、その設定が正しく設定されていることを確認します。

パイプ設定の指定

パイプはデフォルトで有効ですが、無効にすることもできます。パイプのアクセス許可の指定、パイプのロギングの設定、タグの追加を行うこともできます。

パイプ設定を行うには

  1. [Pipe settings] (パイプ設定) タブを選択します。

  2. デフォルトでは、新しく作成されたパイプは作成されるとすぐに有効になります。無効なパイプを作成する場合は、[Activation] (アクティベーション) の [Activate pipe]] (パイプを有効化) で [Active] (有効) をオフにします。

  3. [Permissions] (アクセス許可) の [Execution role] (実行ロール) で、次のいずれかを実行します。

    1. EventBridge にこのパイプに対する新しい実行ロールを作成させるには、[Create a new role for this specific resource] (この特定のリソースに対して新しいロールを作成する) を選択します。[Role name] (ロール名) では、ロール名をオプションで編集できます。

    2. 既存の実行ロールを使用するには、[Use an existing role] (既存のロールを使用する) を選択します。[Logging role] (ログ記録ロール) でロールを選択します。

  4. (オプション) Kinesis または DynamoDB ストリームをパイプソースとして指定すると、再試行ポリシーとデッドレターキュー (DLQ) を設定できます。

    [再試行ポリシーとデッドレターキュー – オプション] で、次の操作を行います。

    [Retry policy] (再試行ポリシー) で、以下の作業を行います。

    1. 再試行ポリシーを有効にする場合は、[Retry] (再試行) をオンにします。デフォルトでは、新しく作成されたパイプは、再試行ポリシーが有効になっていません。

    2. 最大イベント有効期間 に、1 分 (00:01) から 24 時間 (24:00) の間の値を入力します。

    3. 再試行 で、0～185 の数値を入力します。

    4. デッドレターキュー (DLQ) を使用する場合は、[デッドレターキュー] を有効にして、希望する方法を選択し、使用するキューまたはトピックを選択します。デフォルトでは、新規作成したパイプは DLQ を使用しません。

  5. EventBridge がパイプデータの暗号化に使用する KMS key を選択します。

    EventBridge が KMS keys を使用する方法の詳細については、「保管中の暗号化」を参照してください。

    • [AWS 所有のキー を使用] を選択すると、EventBridge は AWS 所有のキー を使用してデータを暗号化します。

      この AWS 所有のキー は、複数の AWS アカウントで使用できるように EventBridge が所有および管理している KMS key です。一般に、リソースを保護する暗号化キーを制御する必要がない限り、AWS 所有のキー は良い選択です。

      これがデフォルトです。

    • [カスタマー管理キー を使用] を選択すると、EventBridge は指定または作成した カスタマー管理キー を使用してデータを暗号化します。

      カスタマーマネージドキー は、お客様が作成、所有、管理している AWS アカウントの KMS keys です。ユーザーは、この KMS keys に関する完全なコントロール権を持ちます。

      1. 既存の カスタマー管理キー を選択するか、[新しい KMS key を作成] を選択します。

        EventBridge は、キーステータスと、指定された カスタマー管理キー に関連付けられているキーエイリアスを表示します。

  6. (オプション) [ログ - オプション] で、EventBridge Pipe から、サポートされているサービスにログ情報を送信する方法 (ログの設定方法を含む) を設定できます。

    パイプレコードのログ記録の詳細については、「Amazon EventBridge Pipes のパフォーマンスのログ記録」を参照してください。

    ERROR ログレベルと同様に、CloudWatch Logs がログの送信先としてデフォルトで選択されます。そのため、デフォルトでは、EventBridge Pipes は新しい CloudWatch ロググループを作成し、このロググループに ERROR レベルの詳細を含むログレコードを送信します。

    EventBridge Pipes から、サポートされているログ送信先のいずれかにログレコードを送信するには、次の操作を行います。

    1. [ログ - オプション] で、ログレコードの配信先を選択します。

    2. [ログレベル] で、ログレコードに含める EventBridge の情報のレベルを選択します。ERROR ログレベルはデフォルトで選択されています。

      詳細については、「EventBridge Pipes のログレベルの指定」を参照してください。

    3. EventBridge でイベントペイロード情報やサービスリクエスト/レスポンス情報をログレコードに含める場合は、[実行データを含める] を選択します。

      詳細については、「EventBridge Pipes ログに実行データを含める」を参照してください。

    4. 選択したログの送信先をそれぞれ設定します。

      CloudWatch Logs ログの場合は、[CloudWatch ログ] で次の操作を行います。

      • [CloudWatch ロググループ] において、EventBridge で新しいロググループを作成するか、既存のロググループを選択する (または既存のロググループの ARN を指定する) かを選択できます。

      • 新しいロググループを作成する場合は、必要に応じてロググループ名を編集します。

      CloudWatch ログはデフォルトで選択されています。

      Firehose ストリームログの場合は、[Firehose ストリームログ] で Firehose ストリームを選択します。

      Amazon S3 ログの場合は、[S3 ログ] で次の操作を行います。

      • ログの送信先として使用するバケットの名前を入力します。

      • バケット所有者の AWS アカウント ID を入力します。

      • EventBridge で S3 オブジェクトの作成時に使用するプレフィックステキストを入力します。

        詳細については、「Amazon Simple Storage Service ユーザーガイド」の「プレフィックスを使用してオブジェクトを整理する」を参照してください。

      • EventBridge で S3 ログレコードをフォーマットする方法を選択します。

  7. (オプション) [Tags - optional] (タグ-オプション) で、[Add new tag] (新しいタグを追加する) を選択し、ルールに対する 1 つまたは複数のタグを入力します。詳細については、「Amazon EventBridge でのリソースのタグ付け」を参照してください。

  8. [Create pipe] (パイプの作成) を選択します。

構成パラメータの検証

パイプが作成されると、EventBridge は次の設定パラメーターを検証します。

  • IAM ロール — パイプの作成後はパイプのソースを変更できないため、EventBridge は指定された IAM ロールがソースにアクセスできることを検証します。

    注記

    EventBridge はエンリッチメントやターゲットに対して同じ検証を行いません。エンリッチメントやターゲットはパイプの作成後に更新される可能性があるためです。

  • バッチ処理 — EventBridge は、ソースのバッチサイズがターゲットの最大バッチサイズを超えていないことを検証します。超えている場合、EventBridge ではバッチサイズを小さくする必要があります。また、ターゲットがバッチ処理をサポートしていない場合、EventBridge でソースのバッチ処理を設定することはできません。

  • エンリッチメント — EventBridge は、API Gateway と API 送信先エンリッチメントのバッチサイズが 1 であることを検証します。これは、バッチサイズ 1 のみがサポートされているためです。