【re:Invent 2022】EventBridge PipesでDDB→Lambdaをしてみた

## はじめに

この記事は、[ミロゴス Advent Calendar 2022](https://qiita.com/advent-calendar/2022/milogos) 2日目の記事です。

re:Inventはオンラインでの参加のため、KeynoteやLeaders Sessionしか見れない側の人間です。
ちょうどアドカレの担当日と、Werner VogelsのKeynoteの日が被っていたのもあり、気になったものをピックアップし、触ってみました。

Keynote冒頭ではレイテンシやスループットの話から非同期の話に繋がり、イベント駆動アーキテクチャに関する話題が前半のメインになりました。
イベント駆動における主役、EventBridgeやStep Functions周りで新サービスが発表され、個人としては非常に興味深いアップデートが多かったです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/a7df0984-8aba-3b3a-7d67-4681136427e8.png)

- Amazon EventBridge Pipes
    -  What's New: [Amazon EventBridge Pipes is now generally available](https://aws.amazon.com/about-aws/whats-new/2022/12/amazon-eventbridge-pipes-generally-available/)
    - AWS News Blog: [New — Create Point-to-Point Integrations Between Event Producers and Consumers with Amazon EventBridge Pipes](https://aws.amazon.com/jp/blogs/aws/new-create-point-to-point-integrations-between-event-producers-and-consumers-with-amazon-eventbridge-pipes/)
- Step Functions Distributed Map
    - What's New: [AWS Step Functions launches large-scale parallel workflows for data processing and serverless applications](https://aws.amazon.com/about-aws/whats-new/2022/12/aws-step-functions-large-scale-parallel-workflows-data-processing-serverless-applications/)
    - AWS News Blog: [Step Functions Distributed Map – A Serverless Solution for Large-Scale Parallel Data Processing](https://aws.amazon.com/jp/blogs/aws/step-functions-distributed-map-a-serverless-solution-for-large-scale-parallel-data-processing/)
- 現時点ではプレビューなサービス
    - Amazon CodeCatalyst
        - What's New: [Announcing Amazon CodeCatalyst (Preview)](https://aws.amazon.com/jp/about-aws/whats-new/2022/12/announcing-amazon-codecatalyst-preview/)
        - AWS News Blog: [Announcing Amazon CodeCatalyst (preview), a Unified Software Development Service](https://aws.amazon.com/jp/blogs/aws/announcing-amazon-codecatalyst-preview-a-unified-software-development-service/)
    - AWS Application Composer。
        - What's New: [Introducing AWS Application Composer (Preview)](https://aws.amazon.com/jp/about-aws/whats-new/2022/12/aws-application-composer-preview/)
        - AWS News Blog: [Visualize and create your serverless workloads with AWS Application Composer](https://aws.amazon.com/jp/blogs/compute/visualize-and-create-your-serverless-workloads-with-aws-application-composer/)
        - AWSアーキテクチャをGUIで設計でき、[SAMを使ってデプロイができるようです](https://twitter.com/edjgeek/status/1598370719045296129)。

ここではこれらの中からEventBridge Pipesについて書いていこうと思います。

## EventBridge Pipes

EventBridgeと言えばサービス間の橋渡し役で、プロデューサーサービスから渡されたイベントをRuleに記載したイベントパターンでフィルタリングし、イベントコンシューマーに渡していました。
EventBridge Pipesでは同じようにプロデューサーとコンシューマー間の橋渡しをしますが、間にフィルタリングやEnrichmentを挟むことでコンシューマーへ渡すイベントを変換できるようです。

それぞれのコンポーネントとして下記のサービスがサポートされているようです。（発表当日の2022/12/2のAWSコンソール上で確認できたものをベースにしています。）

- イベントソース
    - Kinesis
    - SQS
    - DynamoDB
    - MQ
    - MSK
    - self-managed Kafka
- Enrichment
    - Lambda
    - Step Functions
    - API Gateway
    - API Destinations
- ターゲットサービス
    - API Gateway
    - API Destinations
    - Lambda
    - Cloudwatch Logs
    - ECS
    - EventBridge
    - Kinesis Data Firehose
    - Inspector
    - Kinesis Data Streams
    - Redshift
    - SageMaker Pipeline
    - SNS
    - SQS
    - Step Functions
    - Batch

## Pipesを作ってみる

試しにAWSコンソール上で触ってみます。

流れとしてはDynamoDBでのイベントを元に、Filteringを使ってNewImageにidという属性がありeventNameがDelete以外のイベントに限定します。
その後、Enrichmentとターゲットソースには同じLambdaを指定し、それぞれのログを確認できるようにしています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/3a646bba-ccda-a33c-36cb-1883c272a1b2.png)

呼ばれるLambdaは下記のプログラムです。
Enrichmentが受け取る`event`変数の確認と、EnrichmentからターゲットソースのLambdaに渡した値を`event`変数で確認します。

```python:lambda_function.py
import json
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    logger.info(event)
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda through ignore DELETE !')
    }
```

### イベントソース

イベントソースにはDynamoDBを指定します。
なお、DynamoDBはDynamoDB Streamである必要があります。
（選択した後はテーブル名が見えないので、選択ミスにご注意ください。）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/e423805c-f144-1f9d-c66b-e0363a01dae5.png)

今回、DynamoDBにはPartition keyとしてString型のidのみを設定しています。

### フィルタリング

フィルタリングはオプションになります。

コンソール上ではEventBridge Pipesの全体なUIとして、イベントソースのサンプルを確認できます。
そのため、想定しているイベントソースのサンプルを基にフィルタリングの設定ができます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/08caf842-f1e7-c70f-b879-483f5f0adcfc.png)

下記のようなFilteringを設定しています。DELETE以外にしているので、Itemを削除した場合には後ろのLambdaは実行されません。

```json
{
  "dynamodb": {
    "NewImage": {
      "id": {
        "S": [{
          "exists": true
        }]
      }
    }
  },
  "eventName": [{
    "anything-but": "DELETE"
  }]
}
```

なお、フィルタリングの例は下記に記載があります。
- [Amazon EventBridge イベントパターンでのコンテンツのフィルタリング](https://docs.aws.amazon.com/ja_jp/eventbridge/latest/userguide/eb-event-patterns-content-based-filtering.html)

### Enrichment

フィルタリングと同じく、オプション機能です。

Filteringで制限されたイベントを受け取り、そのイベントのJSONをログに出して、固定文字を返すようにしています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/58b735b7-cff8-6126-1a66-beffe5991379.png)

### ターゲットサービス

Enrichmentから渡されるイベントを確認してみます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/6e2c2dc5-d0e7-2e1f-0584-73bd648f4293.png)

### pipesに流してみる

1. DynamoDBにItemを追加してみます。

idは`First_hoge`としてみました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/fd9fa5ab-b5a8-a8c6-3578-69cb45d56b52.png)

2.Lambdaの実行ログを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/224198/b6e1db89-13e0-3835-2a0a-080d26375b5f.png)

Enrichmentとターゲットサービスを同じLambdaにしているため、CloudWatch Logsのログも同じログストリームに出ています。
青枠がEnrichmentのログ、赤枠がターゲットサービスのログです。

Enrichmentのログは下記のような形でした。（JSONの形にフォーマットしています。）
DynamoDB Streamと同じようです。

```json:Enrichmenのevent
[
  {
    "eventID": "e6575fc2c02eb96495d6c3d04a941d8f",
    "eventName": "INSERT",
    "eventVersion": "1.1",
    "eventSource": "aws:dynamodb",
    "awsRegion": "ap-northeast-1",
    "dynamodb": {
      "ApproximateCreationDateTime": 1669946402,
      "Keys": { "id": { "S": "First_hoge" } },
      "NewImage": { "id": { "S": "First_hoge" } },
      "SequenceNumber": "13700000000003743678089",
      "SizeBytes": 24,
      "StreamViewType": "NEW_AND_OLD_IMAGES"
    },
    "eventSourceARN": "arn:aws:dynamodb:ap-northeast-1:0000000000000:table/Pipe-test-table/stream/2022-12-02T01:16:40.686"
  }
]
```

ターゲットサービスのログはEnrichmentで返しているJSONをそのまま受け取っているようです。
```json:ターゲットサービスのイベント
[{'statusCode': 200, 'body': '"Hello from Lambda through ignore DELETE !"'}]
```

## まとめ

re:Inventで発表されたEventBridgeやStep Functions周りの新サービスのうち、EventBridge Pipesを触ってみました。
現在稼働させているプロダクトにて、DynamoDB StreamからLambdaにイベントを流し、Lambdaで条件分岐させるような実装をしているのですが、pipesを使えばコード数がかなり削減できそうでした。