2026.10.17 AWS

EventBridge Pipesの使い方：DynamoDB Streamsのフィルタとエンリッチ実装

Q: パイプが動かないときはどこを確認しますか？

まず aws pipes describe-pipe で CurrentState と StateReason を見ます。権限不足やソースの消失が起きた場合、ここに表示されるのがその理由です。状態が RUNNING なのにターゲットへ届かない場合は、フィルタで落ちているか、エンリッチが空を返している可能性があります。CLIで作ったパイプはログが既定でオフなので、 update-pipe でログレベルをINFOかTRACEに上げてから、もう一度イベントを流します。

Q: CloudFormationやTerraformでも作れますか？

コードによるパイプの作成も可能です。CloudFormationでは AWS::Pipes::Pipe リソースとして定義でき、公式の開始方法のページにDynamoDB StreamsからSQSへ送るテンプレートが載っています。プロパティの名前は本記事のCLIのJSONと同じ構造で、 SourceParameters や FilterCriteria はそのまま移せる項目です。本番では実行ロール、DLQ、ロググループまで同じテンプレートに含め、手作業の設定を残さないようにします。

EventBridge Pipesの使い方：DynamoDB Streamsのフィルタとエンリッチ実装

EventBridge Pipesは、DynamoDB StreamsやSQSのようなストリーム・キューから読んだイベントを、絞り込み、外部データを足し、1つの送信先へ渡すまでをAWS側で受け持つ機能です。これまで「読んで、整形して、送るだけ」のLambdaを書いていた箇所を、設定だけで置き換えられる仕組みです。本記事では、DynamoDBの注文テーブルで支払い済みになった変更だけを拾い、Lambdaで顧客情報を足してイベントバスへ流す構成を、AWS CLIで1つずつ作ります。フィルタの書き方、既定でオフになっているログ、DLQとリトライ、料金の見積もりも2026年10月時点の公式ドキュメントで確かめながら進めます。

まとめ：EventBridge Pipesで橋渡しのLambdaを減らすときの要点

パイプは「ソース→フィルタ→エンリッチ→ターゲット」の4段で動き、フィルタとエンリッチは省略できます。作る前に押さえる点は次の5つです。

1本のパイプはソース1つ・ターゲット1つです。複数の送信先へ配るなら、ターゲットにイベントバスを置き、その先のルールで分けます。
DynamoDBのフィルタはdynamodbフィールドを起点に書きます。eventNameは公式のフィルタ仕様で使えないフィールドに挙がっています。
CLIやIaCで作ったパイプのログは、指定しない限り出ません。作成時に--log-configurationでINFO以上を指定します。
DynamoDBソースのリトライは既定で無期限です。DLQとMaximumRecordAgeInSecondsを24時間より短く設定し、失敗レコードが黙って消えるのを防ぎます。
料金はフィルタ通過後のリクエストだけにかかり、公式の計算例では100万リクエストあたり0.40 USDです。バッチにまとめると請求されるリクエスト数が減ります。

EventBridge Pipesの4段の処理とイベントバス・Lambda直結との違い

手を動かす前に、パイプが何を受け持ち、何を受け持たないかを整理します。

ソース6種とターゲット15種をつなぐフィルタ・エンリッチの4段構成

EventBridge Pipesの公式ガイドは、パイプの作成を「ソースを選ぶ、任意でフィルタを足す、任意でエンリッチを定義する、ターゲットを選ぶ」の順で説明しています。ソースの一覧は、DynamoDB Streams、Kinesis Data Streams、Amazon MQ、Amazon MSK、SQS、Apache Kafkaの6種です。S3やAPI Gatewayからの直接の受信は無く、ストリームかキューを読む専用の部品だと考えると迷いません。

送信先はターゲットの一覧に15種が並び、Lambda、Step Functions、SQS、SNS（FIFOトピックは対象外）、Kinesis、Firehose、ECSタスク、API Gateway、API送信先、同じアカウント・リージョンのイベントバスなどが含まれます。渡せるペイロードは、パイプの上限6MBとターゲット側の上限の小さい方で、イベントバスなら1MB、Step Functionsなら256KBです。エンリッチで情報を足すほど膨らむので、ターゲットの上限から逆算して足す項目を決めます。

多対多のイベントバス・ESM直結と1対1のPipesの役割比較

仕組み	受け取る元	送り先	向く場面
イベントバスとルール	多数：AWSサービス/SaaS/PutEvents	ルールごとに5ターゲット（Classic）	1つのイベントを複数部署へ配る
Pipes	ストリーム・キュー1つ	ターゲット1つ	読んで絞って整えて渡す配管
Lambdaのイベントソースマッピング	ストリーム・キュー1つ	Lambda関数のみ	受け取った先で業務処理を書く

イベントバスの仕組みとルールの書き方はAmazon EventBridgeの仕組みとCLI・Pythonでの実装手順で扱っています。Pipesが置き換えるのは、表の3行目のうち「Lambdaの中身が変換と転送だけ」のケースです。SQSの先がLambdaで、そこで業務処理を書くのであれば、Pipesを挟まずイベントソースマッピングで直結します。その判断の根拠はEventBridgeとSQS・Lambdaの非同期構成で、部分バッチ応答と同時実行の制御の観点から整理しています。

DynamoDB StreamsからPipesでイベントバスへ流すCLIの構築手順

注文テーブルordersの項目がstatus = PAIDになったら、顧客テーブルからメールアドレスを足して、イベントバスorders-busへOrderPaidとして送る構成を作ります。リージョンは東京（ap-northeast-1）、AWS CLIはv2を前提とします。

ストリーム有効化とイベントバス・DLQ用SQSキューを作る準備

# 新旧イメージを流すストリームを付けて注文テーブルを作る
aws dynamodb create-table --table-name orders \
  --attribute-definitions AttributeName=orderId,AttributeType=S \
  --key-schema AttributeName=orderId,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

STREAM_ARN=$(aws dynamodb describe-table --table-name orders \
  --query "Table.LatestStreamArn" --output text)

# 送信先のイベントバス、失敗レコードを退避するDLQ、パイプのロググループ
aws events create-event-bus --name orders-bus
aws sqs create-queue --queue-name orders-pipe-dlq
aws logs create-log-group --log-group-name /aws/vendedlogs/pipes/orders-paid-pipe

ストリームの表示タイプはNEW_AND_OLD_IMAGESにしておくと、変更前後の値をフィルタやエンリッチで比べられます。DynamoDBソースのページに掲載されているのは、ストリームを無効化する際の注意事項です。ソースのストリームを無効化すると、再び有効化してもパイプは使えなくなります。再有効化したストリームには新しいARNが付き、パイプのソースは作成後に変えられないためです。テーブル設定を触る運用手順に「ストリームを止めない」を明記しておきます。

pipes.amazonaws.comが引き受ける実行ロールと4つの読み取り権限

# 信頼ポリシー：パイプのサービスがこのロールを引き受ける
cat > trust.json <<'EOF'
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": { "Service": "pipes.amazonaws.com" },
    "Action": "sts:AssumeRole",
    "Condition": { "StringEquals": { "aws:SourceAccount": "123456789012" } }
  }]
}
EOF

# 権限：ストリームの読み取り、エンリッチ用Lambdaの呼び出し、バスへの送信、DLQへの送信
cat > perm.json <<'EOF'
{
  "Version": "2012-10-17",
  "Statement": [
    { "Effect": "Allow",
      "Action": ["dynamodb:DescribeStream", "dynamodb:GetRecords",
                 "dynamodb:GetShardIterator", "dynamodb:ListStreams"],
      "Resource": "arn:aws:dynamodb:ap-northeast-1:123456789012:table/orders/stream/*" },
    { "Effect": "Allow", "Action": "lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:orders-enrich" },
    { "Effect": "Allow", "Action": "events:PutEvents",
      "Resource": "arn:aws:events:ap-northeast-1:123456789012:event-bus/orders-bus" },
    { "Effect": "Allow", "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:ap-northeast-1:123456789012:orders-pipe-dlq" }
  ]
}
EOF

aws iam create-role --role-name orders-pipe-role --assume-role-policy-document file://trust.json
aws iam put-role-policy --role-name orders-pipe-role --policy-name orders-pipe \
  --policy-document file://perm.json

ストリーム読み取りの4アクションは、ソースのアクセス許可のページに列挙されたものと同じです。パイプはこのロールを使い、pipes.amazonaws.comとしてエンリッチとターゲットを呼び出します。権限が足りないと4xxエラーが続き、パイプは自動で無効化されます。原因の確認先は、後述するStateReasonです。

create-pipeでPAIDの変更だけを通しバッチ10件でバスへ送る指定

cat > source.json <<'EOF'
{
  "DynamoDBStreamParameters": {
    "StartingPosition": "TRIM_HORIZON",
    "BatchSize": 10,
    "MaximumBatchingWindowInSeconds": 5,
    "MaximumRetryAttempts": 5,
    "MaximumRecordAgeInSeconds": 3600,
    "OnPartialBatchItemFailure": "AUTOMATIC_BISECT",
    "DeadLetterConfig": { "Arn": "arn:aws:sqs:ap-northeast-1:123456789012:orders-pipe-dlq" }
  },
  "FilterCriteria": {
    "Filters": [{ "Pattern": "{\"dynamodb\":{\"NewImage\":{\"status\":{\"S\":[\"PAID\"]}},\"OldImage\":{\"status\":{\"S\":[{\"anything-but\":\"PAID\"}]}}}}" }]
  }
}
EOF

cat > target.json <<'EOF'
{
  "EventBridgeEventBusParameters": { "Source": "app.orders", "DetailType": "OrderPaid" }
}
EOF

aws pipes create-pipe --name orders-paid-pipe \
  --role-arn arn:aws:iam::123456789012:role/orders-pipe-role \
  --source "$STREAM_ARN" --source-parameters file://source.json \
  --enrichment arn:aws:lambda:ap-northeast-1:123456789012:function:orders-enrich \
  --target arn:aws:events:ap-northeast-1:123456789012:event-bus/orders-bus \
  --target-parameters file://target.json \
  --log-configuration '{"CloudwatchLogsLogDestination":{"LogGroupArn":"arn:aws:logs:ap-northeast-1:123456789012:log-group:/aws/vendedlogs/pipes/orders-paid-pipe"},"Level":"INFO"}'

フィルタは「変更後がPAIDで、変更前がPAID以外」の更新だけを通します。変更前の条件を外すと、支払い済みの注文で配送先などの別の項目を書き換えるたびに、同じ注文の通知がもう一度流れます。開始位置はTRIM_HORIZONにしました。DynamoDBソースのページによると、パイプの作成直後はポーリングの開始まで数分かかることがあり、LATESTではその間の変更を取りこぼすおそれがあるからです。BatchSizeを10にしたのは、バッチ処理と同時実行のページでイベントバスへ送れる最大バッチが10件と定められているためです。SQSとSNSも10件、KinesisとFirehoseは500件で、ターゲットの上限を超えるバッチサイズは設定できません。

最後の--log-configurationは省かないでください。Pipesのログ記録のページによると、コンソールで作るとCloudWatch LogsへのERRORレベルが既定で選ばれますが、プログラムから作る場合に既定値はありません。CLIで作ったパイプは、指定しなければ何も記録しないということです。INFOなら実行の開始・成功・失敗が残ります。リクエストと応答の中身まで見たいときはIncludeExecutionDataにALLを足しますが、個人情報がログに残るので本番では外します。

put-itemで更新を流しdescribe-pipeとログで通過を確かめる手順

# 状態がRUNNINGになるまで待つ（作成直後はCREATING）
aws pipes describe-pipe --name orders-paid-pipe \
  --query "{state:CurrentState,reason:StateReason}"

# 通らないはずの作成と、通るはずの更新を1件ずつ流す
aws dynamodb put-item --table-name orders --item \
  '{"orderId":{"S":"o-1001"},"customerId":{"S":"c-01"},"status":{"S":"PENDING"},"amount":{"N":"4800"}}'
aws dynamodb update-item --table-name orders --key '{"orderId":{"S":"o-1001"}}' \
  --update-expression "SET #s = :p" --expression-attribute-names '{"#s":"status"}' \
  --expression-attribute-values '{":p":{"S":"PAID"}}'

# 実行の記録を確認する
aws logs tail /aws/vendedlogs/pipes/orders-paid-pipe --since 10m

ログに実行が1回だけ記録されていれば、フィルタは意図どおりです。続けて同じ注文のamountだけを書き換え、実行が増えないことも確かめます。増えた場合はOldImageの条件が効いていません。Patternの文字列はエスケープが多く、引用符の閉じ忘れがあると作成時に例外で止まります。

フィルタ・エンリッチ・入力変換で書くJSONとソース別の注意点

パイプの挙動がおかしいとき、原因の多くはこの3つの設定にあります。

SQSのbody・DynamoDBのdynamodb起点のフィルタとeventName

Pipesのフィルタリングのページでは、本文にあたるフィールドの名前がソースごとに決まっています。SQSはbody、Kinesisはdata、DynamoDBはdynamodbです。SQSでは、フィルタがJSONなのにメッセージが普通の文字列だった場合、そのメッセージは評価されずに捨てられます。

同じページは、ポーリングの過程で付くawsRegion、eventSource、eventNameなど8つのフィールドを、イベントパターンに使えないものとして挙げています。一方で公式の開始方法のチュートリアルは、eventNameがINSERTかMODIFYのものを通すフィルタを例にしています。2026年10月17日時点では、2つのページの記述に食い違いがある状態です。本記事のフィルタがdynamodbの中だけで条件を書いているのはこのためです。仕様ページの側に寄せておけば、どちらに統一されても動きは変わりません。イベントバスのルール側で使える演算子の書き方とテスト方法は、EventBridge イベントパターンの書き方：演算子一覧とtest-event-patternでの検証にまとめています。

SQSソースには別の落とし穴があります。フィルタに一致しなかったメッセージは、パイプがキューから自動で削除します。1つのキューに複数のパイプをつなぐと、互いに相手向けのメッセージを消し合うため、公式は推奨していません。用途ごとにキューを分けてください。

Lambdaエンリッチで顧客情報を足す同期呼び出しと空応答の扱い

import boto3

customers = boto3.resource("dynamodb").Table("customers")


def handler(records, context):
    # パイプからはバッチ（最大10件）がリストで渡される
    out = []
    for r in records:
        new = r["dynamodb"]["NewImage"]
        item = customers.get_item(Key={"customerId": new["customerId"]["S"]}).get("Item")
        if item is None or item.get("unsubscribed"):
            continue  # 通知しない注文はここで落とす
        out.append({
            "orderId": new["orderId"]["S"],
            "amount": int(new["amount"]["N"]),
            "email": item["email"],
            "eventID": r["eventID"],
        })
    return out  # 空リストを返すとターゲットは呼ばれない

エンリッチメントのページによると、エンリッチは応答を待ってからターゲットを呼ぶ同期呼び出しで、応答は6MBまでです。使えるのはLambda、API Gateway、API送信先、Step FunctionsのExpressワークフローで、Standardワークフローは置けません。空の文字列、{}、[]を返すとターゲットは呼ばれないため、上のコードのように宛先の無い注文をエンリッチ側で落とせます。空のペイロードのままターゲットを呼びたいときは[{}]を返します。

戻り値にソースのeventIDを残しているのは、ターゲットがLambdaやStep Functionsのとき部分バッチ失敗を返せるようにするためです。バッチ処理のページは、エンリッチが返す配列にこの識別子を含める必要があると書いています。なお、エンリッチ自体の部分バッチ失敗には対応していません。エンリッチで1件でも例外を出すと、バッチ全体がリトライされます。

InputTemplateで各レコードを整形する入力変換と予約変数の使い方

エンリッチが要らず、項目を抜き出して並べ替えるだけなら、Lambdaを書かずに入力変換で済みます。入力変換のページの書式で、target.jsonに次のInputTemplateを足すと、バスのdetailに入る中身を整えられます。

{
  "EventBridgeEventBusParameters": { "Source": "app.orders", "DetailType": "OrderPaid" },
  "InputTemplate": "{\"orderId\": \"<$.dynamodb.NewImage.orderId.S>\", \"amount\": \"<$.dynamodb.NewImage.amount.N>\",\"pipe\": \"<aws.pipes.pipe-name>\", \"receivedAt\": \"<aws.pipes.event.ingestion-time>\"}"
}

入力変換は、バッチの配列全体ではなく配列の1レコードずつに適用されます。値を差し込む変数には引用符を付け、オブジェクトや配列を差し込む変数には付けないのが決まりです。DynamoDBの数値型Nは文字列として届くため、amountも文字列として渡し、数値への変換は受け取る側で行います。エンリッチを置いたパイプでは、ターゲット側の入力変換が受け取るのはエンリッチの戻り値で、ソースのdynamodbフィールドは残っていません。その場合はパスを<$.orderId>のように戻り値の形に合わせます。予約変数には、パイプ名の<aws.pipes.pipe-name>や受信時刻の<aws.pipes.event.ingestion-time>などがあり、後段で「どのパイプを通ったか」を追うのに使えます。ルールのターゲットに掛ける入力トランスフォーマーとの違いはEventBridge 入力トランスフォーマーの使い方で解説しています。

リトライ・DLQ・同時実行の上限と料金64KB単位での見積もり方

本番で問題になるのは、正常系よりも失敗したレコードの行き先と、件数が増えたときの上限です。

DynamoDBソースのDLQと最大レコード経過時間を24時間未満に置く理由

AWS CLIのcreate-pipeリファレンスでは、MaximumRetryAttemptsとMaximumRecordAgeInSecondsの既定値はどちらも-1、つまり無期限です。何も指定しないと、失敗し続けるレコードはストリームの保持期限まで同じシャードの後続を止め、期限が来た時点で消えます。DynamoDBのストリームの保持は通常24時間です。

エラー処理のページは、ストリームソースのDLQにSQSキューかSNSトピックを指定でき、MaximumRecordAgeInSecondsはソースの保持期間より短くするよう求めています。DLQへ送られてから原本が消えるまでの間に、原因を調べる時間を残すためです。先のsource.jsonで3600秒（1時間）とリトライ5回を指定したのはこの理由によります。FIFOキューはストリームソースのDLQに使えません。SQSソースの場合はパイプ側にDLQ設定は無く、キューに設定したDLQと最大受信数がそのまま使われます。

失敗が続くパイプを見つける手掛かりは、パイプの状態です。describe-pipeのStateReasonには「Stream not found」や権限不足の文言が入り、権限やリソースの欠落による4xxが続くとパイプは自動で無効化されます。CloudWatchアラームを掛けるなら、この状態の変化とDLQのメッセージ数の2つを監視対象にします。

1実行5分の時間制限と東京リージョン同時実行1,000の上限

バッチ処理のページによると、1回の実行はエンリッチとターゲットの処理を合わせて5分までで、この上限は引き上げられません。エンリッチで外部APIを1件ずつ呼ぶ設計は、バッチ10件でも遅いAPIなら5分に届きます。重い処理はエンリッチではなく、ターゲットのStep Functionsやキューの先に回します。AWS Step Functionsのステートマシンと2種のワークフローで触れているとおり、Standardワークフローは非同期でしか呼べず、パイプはその中のエラーを検知してリトライできません。

EventBridgeのクォータでは、アカウントあたりの同時実行は東京リージョンで1,000、バージニア北部・オレゴン・アイルランドで3,000、パイプの数は全リージョンで1,000が既定です。DynamoDBソースの同時実行はシャード数×ParallelizationFactor（1〜10）まで伸び、順序はパーティションキーの単位で守られます。SQS標準キューは1本のパイプで1,250まで伸びるため、アカウント全体の1,000を1本で使い切ることもあり得ます。

フィルタ通過後100万リクエスト0.40ドルとバッチで請求を減らす試算

Amazon EventBridgeの料金ページの計算例では、SQSが月に1,000万件を受け取り、フィルタに25％が一致し、5件ずつバッチにまとめると、請求されるリクエストは50万件です。100万リクエストあたり0.40 USDで、月0.20 USDになります。ペイロードは64KBごとに1リクエストとして数えられます。

この例から読み取れるのは、バッチにまとめた単位で数えられるという点です。注文の変更が月に1,000万件あり、PAIDへの更新が10％、バッチ10件なら、請求は10万リクエストで月0.04 USDです。エンリッチのLambdaとイベントバスの料金は別にかかります。バスへの送信を含む全体の単価と、請求が跳ねる5つのパターンはAmazon EventBridgeの料金と64KB課金の落とし穴にまとめています。

橋渡しのLambdaをPipesに置き換える採用条件と見送る場面

ここまでの仕様を踏まえ、置き換えてよい場面とやめておく場面を分けます。

読んで絞って送るだけのLambdaが複数ある構成での採用条件

採用してよいのは、既存のLambdaの中身が「ストリームやキューを読み、条件で絞り、形を変えて別サービスへ送る」だけで、業務ロジックを持たないときです。そうした関数が3本、5本と増えている構成では、ランタイムの更新、依存ライブラリの脆弱性対応、同時実行の監視がすべて関数の本数分だけ発生します。パイプに移すと、その保守は設定ファイルのレビューに置き換わります。DynamoDBの変更をイベントとして社内に配りたい場合も、ソースをパイプ、配布をイベントバスのルールに分ける形が公式の想定どおりの組み方です。変更データの取り出し方そのものの比較はチェンジフィードとは：DynamoDB・Cosmos DBでの変更フィード実装で扱っています。

既存のLambda群のうちどれをパイプに移せるかの棚卸しから、IAM・DLQ・監視まで含めた構成の見直しを相談したい場合は、一創のAWS・Google Cloud・Azureのインフラ構築で、現行構成の調査から対応しています。

業務処理を書くLambdaや多数への配信でPipesを見送る判断基準

見送るのは次の3つの場合です。第一に、SQSやストリームの先で業務処理を書くLambdaです。この場合パイプを挟んでも料金と設定が1段増えるだけで、イベントソースマッピングの部分バッチ応答と最大同時実行数のほうが細かく制御できます。第二に、エンリッチで外部APIを何本も呼ぶ処理です。1実行5分の上限と、エンリッチの部分バッチ失敗が非対応である点から、失敗時の再処理範囲が広がります。第三に、ストリームの設定変更が頻繁なテーブルです。ストリームを一度無効化するとパイプを作り直すことになり、その間の変更の扱いを毎回設計する手間が生じます。

EventBridge Pipesを導入する前によくある質問

Pipesを使い始めるときに出やすい疑問に、公式ドキュメントの記載をもとに答えます。

EventBridge PipesとEventBridgeのイベントバスはどう違いますか？

イベントバスは、多数の送信元から来るイベントをルールで多数の送信先へ振り分ける仕組みです。Pipesは、DynamoDB StreamsやSQSなど1つのソースを読み、フィルタとエンリッチを経て1つのターゲットへ渡す1対1の仕組みです。2つは競合せず、パイプのターゲットにイベントバスを置き、その先のルールで複数の送信先へ広げる組み合わせが公式でも代表例として挙がっています。

1本のパイプで複数のターゲットへ送れますか？

送れません。ターゲットはパイプごとに1つです。複数の送信先が必要なら、ターゲットをイベントバスかSNSトピックにして、その先で分岐させます。同じソースに複数のパイプをつなぐ方法は、SQSソースでは不一致のメッセージを互いに削除し合うため推奨されていません。DynamoDB Streamsの解説は、同じシャードを同時に読むプロセスを2つまでとし、超えるとスロットリングが起きうると書いています。パイプを増やす前にバス経由の分岐を検討します。

EventBridge Pipesの料金はいくらですか？

公式の料金ページの計算例では、フィルタを通過したリクエスト100万件あたり0.40 USDです。フィルタで落ちたイベントは数えられず、バッチにまとめた単位で1リクエストになり、ペイロードは64KBごとに1リクエストです。エンリッチに使うLambdaの実行料金、ターゲット側の料金は別にかかります。単価は改定されることがあるので、見積もりの最後に料金ページで確認してください。

パイプが動かないときはどこを確認しますか？

まずaws pipes describe-pipeでCurrentStateとStateReasonを見ます。権限不足やソースの消失が起きた場合、ここに表示されるのがその理由です。状態がRUNNINGなのにターゲットへ届かない場合は、フィルタで落ちているか、エンリッチが空を返している可能性があります。CLIで作ったパイプはログが既定でオフなので、update-pipeでログレベルをINFOかTRACEに上げてから、もう一度イベントを流します。

CloudFormationやTerraformでも作れますか？

コードによるパイプの作成も可能です。CloudFormationではAWS::Pipes::Pipeリソースとして定義でき、公式の開始方法のページにDynamoDB StreamsからSQSへ送るテンプレートが載っています。プロパティの名前は本記事のCLIのJSONと同じ構造で、SourceParametersやFilterCriteriaはそのまま移せる項目です。本番では実行ロール、DLQ、ロググループまで同じテンプレートに含め、手作業の設定を残さないようにします。