DynamoDB Accelerator (DAX)でのトランザクション処理について検証してみた

.
├── README.md
├── __init__.py
├── functions
│   ├── ecommerce_test
│   │   └── app.py
│   ├── transact_get_test
│   │   └── app.py
│   └── transact_write_test
│       └── app.py
├── layers
│   └── amazon-dax-client
│       └── requirements.txt
├── samconfig.toml
└── template.yaml

AWSTemplateFormatVersion: 2010-09-09
Transform: AWS::Serverless-2016-10-31
Description: DAX Transaction Test

Parameters:
  Project:
    Type: String
    Default: dax-transaction-test

Globals:
  Function:
    Timeout: 30
    MemorySize: 256
    Runtime: python3.12
    Layers:
        - !Ref DaxClientLayer
    Environment:
      Variables:
        DAX_ENDPOINT: !GetAtt DaxCluster.ClusterDiscoveryEndpoint
    VpcConfig:
      SecurityGroupIds:
        - !GetAtt DaxSecurityGroup.GroupId
      SubnetIds:
        - !Ref Subnet1
        - !Ref Subnet2

Resources:
  ## Layer
  DaxClientLayer:
    Type: AWS::Serverless::LayerVersion
    Properties:
      LayerName: !Sub "${Project}-dax-client-layer"
      ContentUri: layers/amazon-dax-client/
      CompatibleArchitectures:
        - x86_64
      CompatibleRuntimes:
        - python3.12
    Metadata:
      BuildMethod: python3.12

  # DynamoDBテーブル
  ProductsTable:
    Type: AWS::DynamoDB::Table
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      TableName: !Sub "${Project}-products"
      BillingMode: PROVISIONED
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5
      AttributeDefinitions:
        - AttributeName: ProductId
          AttributeType: S
      KeySchema:
        - AttributeName: ProductId
          KeyType: HASH
      Tags:
        - Key: Project
          Value: !Ref Project

  OrdersTable:
    Type: AWS::DynamoDB::Table
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      TableName: !Sub "${Project}-orders"
      BillingMode: PROVISIONED
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5
      AttributeDefinitions:
        - AttributeName: OrderId
          AttributeType: S
      KeySchema:
        - AttributeName: OrderId
          KeyType: HASH
      Tags:
        - Key: Project
          Value: !Ref Project

  # VPC設定（DAXに必要）
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags:
        - Key: Project
          Value: !Ref Project
        - Key: Name
          Value: !Sub "${Project}-vpc"

  # サブネットx2
  Subnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.0.0.0/24
      AvailabilityZone: !Select [0, !GetAZs ""]
      Tags:
        - Key: Project
          Value: !Ref Project
        - Key: Name
          Value: !Sub "${Project}-subnet-1"

  Subnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.0.1.0/24
      AvailabilityZone: !Select [1, !GetAZs ""]
      Tags:
        - Key: Project
          Value: !Ref Project
        - Key: Name
          Value: !Sub "${Project}-subnet-2"

  # DAX関連リソース
  DaxSubnetGroup:
    Type: AWS::DAX::SubnetGroup
    Properties:
      SubnetGroupName: !Sub "${Project}-subnet-group"
      SubnetIds:
        - !Ref Subnet1
        - !Ref Subnet2

  DaxSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group for DAX cluster
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 8111
          ToPort: 8111
          CidrIp: 10.0.0.0/16
      Tags:
        - Key: Project
          Value: !Ref Project
        - Key: Name
          Value: !Sub "${Project}-sg"

  DaxServiceRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "${Project}-dax-role"
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: dax.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess
      Tags:
        - Key: Project
          Value: !Ref Project

  DaxCluster:
    Type: AWS::DAX::Cluster
    Properties:
      ClusterName: !Sub "test-cluster"
      NodeType: dax.t3.small
      ReplicationFactor: 2
      IAMRoleARN: !GetAtt DaxServiceRole.Arn
      SecurityGroupIds:
        - !GetAtt DaxSecurityGroup.GroupId
      SubnetGroupName: !Ref DaxSubnetGroup

  # Lambda用IAMロール
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "${Project}-lambda-role"
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
        - arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess
      Tags:
        - Key: Project
          Value: !Ref Project

  # Lambda関数
  TransactWriteTestFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub "${Project}-write"
      CodeUri: functions/transact_write_test/
      Handler: app.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Environment:
        Variables:
          PRODUCTS_TABLE: !Ref ProductsTable
      Tags:
        Project: !Ref Project

  TransactGetTestFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub "${Project}-get"
      CodeUri: functions/transact_get_test/
      Handler: app.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Environment:
        Variables:
          PRODUCTS_TABLE: !Ref ProductsTable
      Tags:
        Project: !Ref Project

  EcommerceTestFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub "${Project}-ecommerce"
      CodeUri: functions/ecommerce_test/
      Handler: app.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Environment:
        Variables:
          PRODUCTS_TABLE: !Ref ProductsTable
          ORDERS_TABLE: !Ref OrdersTable
      Tags:
        Project: !Ref Project

Outputs:
  DaxEndpoint:
    Value: !GetAtt DaxCluster.ClusterDiscoveryEndpoint

  ProductsTable:
    Value: !Ref ProductsTable

  OrdersTable:
    Value: !Ref OrdersTable

AvailabilityZone: !Select [1, !GetAZs ""]

import os
import time
from amazondax import AmazonDaxClient

# 環境変数から取得
DAX_ENDPOINT = os.environ['DAX_ENDPOINT']
PRODUCTS_TABLE = os.environ['PRODUCTS_TABLE']

# DAXクライアント
dax = AmazonDaxClient(endpoint_url=DAX_ENDPOINT)

def handler(event, context):
    # 書き込むテストデータ
    products = [
        {
            'ProductId': {'S': 'dax-phone'},
            'Name': {'S': 'スマートフォン'},
            'Price': {'N': '60000'},
            'Stock': {'N': '30'}
        },
        {
            'ProductId': {'S': 'dax-laptop'},
            'Name': {'S': 'ノートパソコン'},
            'Price': {'N': '150000'},
            'Stock': {'N': '15'}
        }
    ]

    # DAX 経由で TransactWriteItems を実行
    dax.transact_write_items(
        TransactItems=[
            {'Put': {'TableName': PRODUCTS_TABLE, 'Item': products[0]}},
            {'Put': {'TableName': PRODUCTS_TABLE, 'Item': products[1]}}
        ]
    )

    # DAX経由のGetItem　を実行
    for p in products:
        key = {'ProductId': p['ProductId']}
        result = dax.get_item(TableName=PRODUCTS_TABLE, Key=key)
        print(f"DAX Cache Read: {result.get('Item')}")

import os
import time
from amazondax import AmazonDaxClient

# 環境変数から取得
DAX_ENDPOINT = os.environ['DAX_ENDPOINT']
PRODUCTS_TABLE = os.environ['PRODUCTS_TABLE']

# DAXクライアント
dax = AmazonDaxClient(endpoint_url=DAX_ENDPOINT)

def handler(event, context):
    # 書き込むテストデータ
    products = [
        {
            'ProductId': {'S': 'dax-desktop'},
            'Name': {'S': 'デスクトップ'},
            'Price': {'N': '120000'},
            'Stock': {'N': '30'}
        },
        {
            'ProductId': {'S': 'dax-filter'},
            'Name': {'S': 'フィルター'},
            'Price': {'N': '10000'},
            'Stock': {'N': '5'}
        }
    ]

    # DAX 経由で TransactWriteItems を実行
    dax.transact_write_items(
        TransactItems=[
            {'Put': {'TableName': PRODUCTS_TABLE, 'Item': products[0]}},
            {'Put': {'TableName': PRODUCTS_TABLE, 'Item': products[1]}}
        ]
    )

    # バックグラウンドキャッシュ更新を待つ（例：1秒）
    time.sleep(1)

    # キャッシュ効果を検証：DAX経由のGetItemを実行
    for p in products:
        key = {'ProductId': p['ProductId']}
        result = dax.get_item(TableName=PRODUCTS_TABLE, Key=key)
        print(f"DAX Cache Read: {result.get('Item')}")

import os
from amazondax import AmazonDaxClient

# 環境変数
DAX_ENDPOINT = os.environ.get('DAX_ENDPOINT')
PRODUCTS_TABLE = os.environ.get('PRODUCTS_TABLE', 'Products')

# DAXクライアント初期化
dax = AmazonDaxClient(endpoint_url=DAX_ENDPOINT)

def handler(event, context):

    # DAX経由でトランザクション読み取り実行
    # キャッシュをバイパスしてDynamoDBから直接取得になるはず
    dax.transact_get_items(
        TransactItems=[
            {
                'Get': {
                    'TableName': PRODUCTS_TABLE,
                    'Key': {'ProductId': {'S': 'dax-phone'}}
                }
            },
            {
                'Get': {
                    'TableName': PRODUCTS_TABLE,
                    'Key': {'ProductId': {'S': 'dax-laptop'}}
                }
            }
        ]
    )