2023年1月現在、DynamoDBのレコード数を確認する
CloudWatchメトリクスがないためこの記事を投稿しました。

全体の流れ

構築するリソース

・DynamoDB

適当のもの

・Lambda

ランタイム： Python3.9

・EventBridge

1分ごとにLambdaをトリガーするスケジュール

・IAMロール/IAMポリシー

Lambda実行ロールとEventBridgeがLambdaを呼び出すロール

0. まずはじめに

手順1以降のコピー&ペーストを行いやすいように、AWSアカウントIDを変数に割り当てる

CLI

export AWS_ACCOUNT_ID=`aws sts get-caller-identity --query Account --output text`

1. 構築 - DynamoDB

適当なもの

1-1. テーブルの作成

CLI

aws dynamodb create-table \
    --table-name TestTable \
    --attribute-definitions AttributeName=id,AttributeType=S AttributeName=name,AttributeType=S \
    --key-schema AttributeName=id,KeyType=HASH AttributeName=name,KeyType=RANGE \
    --provisioned-throughput ReadCapacityUnits=1,WriteCapacityUnits=1

2. 構築 - Lambda

ランタイム： Python3.9

2-1. Lambda実行用IAMロールの作成

CLI

aws iam create-role \
  --role-name TestLambdaRole \
  --assume-role-policy-document \
'{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}'

2-2. Lambda実行用IAMポリシーの作成

CLI

aws iam create-policy \
  --policy-name TestLambdaPolicy \
  --policy-document \
'{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "cloudwatch:PutMetricData",
                "dynamodb:DescribeTable",
                "dynamodb:Scan"
            ],
            "Resource": "*"
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:ap-northeast-1:'${AWS_ACCOUNT_ID}':*"
        },
        {
            "Sid": "VisualEditor2",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:ap-northeast-1:'${AWS_ACCOUNT_ID}':log-group:/aws/lambda/TestLambdaFunction:*"
        }
    ]
}'

2-3. IAMロールへのIAMポリシーのアタッチ

CLI

aws iam attach-role-policy \
  --role-name TestLambdaRole \
  --policy-arn arn:aws:iam::$AWS_ACCOUNT_ID:policy/TestLambdaPolicy

2-4. コードを記述したindex.pyの作成

下記のようなindex.pyファイルを作成
後述する「DescribeTableAPIを使う場合」と「scanAPIを使う場合」の2種類の例

index.py - DescribeTableAPIを使う場合

import json
import boto3

TABLE_NAME = 'TestTable'

dynamodb = boto3.client('dynamodb')
cloudwatch = boto3.client('cloudwatch')

#DescribeTableAPIを使用してレコード数を取得する関数
def describe_table_api():
    response = dynamodb.describe_table(TableName=TABLE_NAME)
    return response['Table']['ItemCount']

def lambda_handler(event, context):
    RecordCount = describe_table_api()
    #メトリクスを発行
    cloudwatch.put_metric_data(
    MetricData = [
        {
            'MetricName': 'DynamoDB_Record_Count',
            'Dimensions': [
                {
                    'Name': 'DynamoDB',
                    'Value': 'RecordCount'
                },
            ],
            'Unit': 'Count',
            'Value': int(RecordCount)
        },
    ],
    Namespace = 'DynamoDBCustomMetrics'
    )
    print(f"Create Metrics Success. Count: {RecordCount}")

index.py - scanAPIを使う場合

import json
import boto3

TABLE_NAME = 'TestTable'

dynamodb = boto3.client('dynamodb')
cloudwatch = boto3.client('cloudwatch')

#scanAPIを使用してレコード数を取得する関数
def scan_api():
    EvaluatedKey = ''
    total_record_count = 0
    while True:
        if EvaluatedKey == '':
            response = dynamodb.scan(
                TableName=TABLE_NAME,
                Select='COUNT'
                )
        else:
            response = dynamodb.scan(
                TableName=TABLE_NAME,
                Select='COUNT',
                ExclusiveStartKey=EvaluatedKey
                )
        total_record_count = total_record_count + response['Count']
        #LastEvaluatedKeyがある場合は、LastEvaluatedKeyが出なくなるまでカウントを足し算
        if 'LastEvaluatedKey' in response:
            EvaluatedKey = response['LastEvaluatedKey']
        else:
            break
    return total_record_count

def lambda_handler(event, context):
    RecordCount = scan_api()
    #メトリクスを発行
    cloudwatch.put_metric_data(
    MetricData = [
        {
            'MetricName': 'DynamoDB_Record_Count',
            'Dimensions': [
                {
                    'Name': 'DynamoDB',
                    'Value': 'RecordCount'
                },
            ],
            'Unit': 'Count',
            'Value': int(RecordCount)
        },
    ],
    Namespace = 'DynamoDBCustomMetrics'
    )
    print(f"Create Metrics Success. Count: {RecordCount}")

2-5. デプロイ用パッケージを作成

CLI

zip index.zip index.py

2-6. Lambdaの作成

CLI

aws lambda create-function \
    --function-name TestLambdaFunction \
    --zip-file fileb://index.zip \
    --handler index.lambda_handler \
    --runtime python3.9 \
    --role arn:aws:iam::$AWS_ACCOUNT_ID:role/TestLambdaRole

3. 構築 - EventBridge

1分ごとにLambdaをトリガーするスケジュール

3-1. EventBridgeトリガー用IAMロールの作成

CLI

aws iam create-role \
  --role-name TestEventbridgeRole \
  --assume-role-policy-document \
'{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "scheduler.amazonaws.com"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringEquals": {
                    "aws:SourceAccount": "'${AWS_ACCOUNT_ID}'",
                    "aws:SourceArn": "arn:aws:scheduler:ap-northeast-1:'${AWS_ACCOUNT_ID}':schedule/default/TestSchedule"
                }
            }
        }
    ]
}'

3-2. EventBridgeトリガー用IAMポリシーの作成

CLI

aws iam create-policy \
  --policy-name TestEventbridgePolicy \
  --policy-document \
'{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": [
                "arn:aws:lambda:ap-northeast-1:'${AWS_ACCOUNT_ID}':function:TestLambdaFunction:*",
                "arn:aws:lambda:ap-northeast-1:'${AWS_ACCOUNT_ID}':function:TestLambdaFunction"
            ]
        }
    ]
}'

3-3. IAMロールへのIAMポリシーのアタッチ

CLI

aws iam attach-role-policy \
  --role-name TestEventbridgeRole \
  --policy-arn arn:aws:iam::$AWS_ACCOUNT_ID:policy/TestEventbridgePolicy

3-4. EventBridgeスケジュールの作成

1分ごとにLambdaをトリガーするスケジュール

CLI

aws scheduler create-schedule \
    --flexible-time-window Mode=OFF \
    --name TestSchedule \
    --schedule-expression 'rate(1 minutes)' \
    --target \
'{
    "RoleArn": "arn:aws:iam::'${AWS_ACCOUNT_ID}':role/TestEventbridgeRole", 
    "Arn":"arn:aws:lambda:ap-northeast-1:'${AWS_ACCOUNT_ID}':function:TestLambdaFunction"
}'

下記エラーが出る場合には、直前に作成したIAMロールのAssumeRole(信頼されたエンティティ)に誤りがある可能性があります。

An error occurred (ValidationException) when calling the CreateSchedule operation: The execution role you provide must allow AWS EventBridge Scheduler to assume the role.

以上！

解説 - テーブルからレコード数を取得する

DynamoDBのレコード数を取得する方法は2種類あります。

DescribeTable APIを使用する
scan APIを使用する

それぞれのメリット・デメリットは、

DescribeTable API

メリット・・・手軽に取得できる。
デメリット・・・リアルタイムでのレコード数の取得は不可。約6時間毎更新。

6時間毎に更新する旨はCLIリファレンスに記載されています。

ItemCount -> (long)
The number of items in the specified table. DynamoDB updates this value approximately every six hours. Recent changes might not be reflected in this value.

指定されたテーブルのアイテム数。DynamoDBはこの値をおよそ6時間ごとに更新します。最近の変更はこの値には反映されないかもしれません。
※DeepL翻訳

scan API

メリット・・・リアルタイムでレコード数の取得が可能。
デメリット・・・読み込みキャパシティユニット（RCU）を消費してしまう。scan APIには1回呼び出しあたり最大1MBに制限される。

scan APIのクォータ（上限）はAWS公式ドキュメントに記載されています。

Scan
Scan の結果セットは、1 回の呼び出しあたり 1 MB に制限されます。スキャン応答から LastEvaluatedKey を使用して、結果をさらに取り出すこともできます。

終わりに

今回は、DynamoDBのテーブルのレコード数をカスタムメトリクスとして発行する方法をまとめました。
カスタムメトリクスとして管理することで、
CloudWatchアラームにて指定の閾値に達したことをトリガーに、
通知や別のプログラムの起動等が可能となります。
レコード数が標準メトリクスとしてリリースされましたら、その記事も執筆したいと思います。
新米エンジニアの初執筆のブログですが、どなたかの役に立てば嬉しいです。