mybook

データストア — DynamoDB と Aurora Serverless

「データはどこに入れるの?」

これはダイチが最初に詰まった問いだった。LambdaはステートレスなのでRailsのようにActiveRecordを使えない。サーバーレスアーキテクチャでのデータ永続化をどう設計するか。

サーバーレスデータベースの選択肢

Loading diagram...
DynamoDBAurora Serverless v2
料金体系リクエスト数 + ストレージACU時間 + ストレージ
スケーリング即座・自動数秒（0.5 ACU単位）
最小コストほぼ0（無料枠あり）~$0.06/ACU-hour
クエリ柔軟性キー・インデックスのみ完全なSQL
JOIN不可可能
トランザクションあり（制限付き）完全なACID

ダイチのECサイトでは、商品データはDynamoDB、注文データはAurora Serverlessという使い分けにした。

DynamoDB の基礎

データモデルの考え方

RDSのテーブル設計は「正規化」が基本だが、DynamoDBは「アクセスパターン駆動」で設計する。

まずアクセスパターンを列挙する:

1. 商品一覧を取得（カテゴリ別、ページネーション）
2. 商品詳細を取得（ID指定）
3. 特定ブランドの商品を取得
4. 価格帯で商品を絞り込む
5. 出品者ごとの商品一覧を取得
Loading diagram...

テーブル定義

# CloudFormation / SAM
ProductsTable:
  Type: AWS::DynamoDB::Table
  Properties:
    BillingMode: PAY_PER_REQUEST
    TableName: !Sub "${AWS::StackName}-products"
    AttributeDefinitions:
      - AttributeName: PK
        AttributeType: S
      - AttributeName: SK
        AttributeType: S
      - AttributeName: category_price
        AttributeType: S
      - AttributeName: brand_id
        AttributeType: S
      - AttributeName: created_at
        AttributeType: S
    KeySchema:
      - AttributeName: PK
        KeyType: HASH
      - AttributeName: SK
        KeyType: RANGE
    GlobalSecondaryIndexes:
      - IndexName: CategoryPriceIndex
        KeySchema:
          - AttributeName: category_price
            KeyType: HASH
          - AttributeName: SK
            KeyType: RANGE
        Projection:
          ProjectionType: ALL
      - IndexName: BrandIndex
        KeySchema:
          - AttributeName: brand_id
            KeyType: HASH
          - AttributeName: created_at
            KeyType: RANGE
        Projection:
          ProjectionType: ALL
    PointInTimeRecoverySpecification:
      PointInTimeRecoveryEnabled: true

データの書き込みと読み込み

# frozen_string_literal: true
 
require 'aws-sdk-dynamodb'
 
class ProductRepository
  TABLE_NAME = ENV['PRODUCTS_TABLE_NAME']
 
  def initialize
    @client = Aws::DynamoDB::Client.new
  end
 
  # 商品の作成
  def create(product_attrs)
    item = {
      'PK' => "PRODUCT##{product_attrs[:id]}",
      'SK' => 'PRODUCT',
      'id' => product_attrs[:id],
      'name' => product_attrs[:name],
      'price' => product_attrs[:price],
      'category' => product_attrs[:category],
      'brand_id' => product_attrs[:brand_id],
      'seller_id' => product_attrs[:seller_id],
      'stock' => product_attrs[:stock],
      'status' => 'active',
      # GSI 用の複合キー: カテゴリ + 価格（ゼロ埋めでソート可能にする）
      'category_price' => "#{product_attrs[:category]}##{product_attrs[:price].to_s.rjust(10, '0')}",
      'created_at' => Time.now.iso8601,
      'updated_at' => Time.now.iso8601
    }
 
    @client.put_item(
      table_name: TABLE_NAME,
      item:,
      # 既存アイテムが無い場合のみ作成
      condition_expression: 'attribute_not_exists(PK)'
    )
 
    item
  end
 
  # ID で商品を取得
  def find_by_id(product_id)
    result = @client.get_item(
      table_name: TABLE_NAME,
      key: {
        'PK' => "PRODUCT##{product_id}",
        'SK' => 'PRODUCT'
      }
    )
 
    result.item
  end
 
  # カテゴリで商品一覧取得（価格でソート）
  def find_by_category(category:, min_price: nil, max_price: nil, limit: 20, last_key: nil)
    prefix = "#{category}#"
    key_condition = 'category_price = :category'
    attr_values = { ':category' => category }
 
    # 価格範囲フィルタ（SKが price なのでbetween使用）
    if min_price || max_price
      key_condition = 'category_price BETWEEN :min AND :max'
      attr_values = {
        ':min' => "#{prefix}#{min_price.to_i.to_s.rjust(10, '0')}",
        ':max' => "#{prefix}#{max_price.to_i.to_s.rjust(10, '0')}"
      }
    end
 
    params = {
      table_name: TABLE_NAME,
      index_name: 'CategoryPriceIndex',
      key_condition_expression: key_condition,
      expression_attribute_values: attr_values,
      limit:,
      scan_index_forward: true  # 価格昇順
    }
 
    params[:exclusive_start_key] = JSON.parse(last_key) if last_key
 
    result = @client.query(params)
 
    {
      items: result.items,
      last_evaluated_key: result.last_evaluated_key ? JSON.generate(result.last_evaluated_key) : nil
    }
  end
 
  # 在庫更新（楽観的ロック）
  def update_stock(product_id:, quantity_delta:, expected_stock:)
    @client.update_item(
      table_name: TABLE_NAME,
      key: {
        'PK' => "PRODUCT##{product_id}",
        'SK' => 'PRODUCT'
      },
      update_expression: 'SET stock = stock + :delta, updated_at = :now',
      condition_expression: 'stock = :expected AND stock + :delta >= :zero',
      expression_attribute_values: {
        ':delta' => quantity_delta,
        ':expected' => expected_stock,
        ':now' => Time.now.iso8601,
        ':zero' => 0
      }
    )
  rescue Aws::DynamoDB::Errors::ConditionalCheckFailedException
    raise StockConflictError, "Stock conflict for product #{product_id}"
  end
end

WARNING

DynamoDBのページネーションはオフセットではなく LastEvaluatedKey を使う。「3ページ目にジャンプ」はできない。UX設計では「次のページ」ボタンを使う無限スクロールパターンが適している。

トランザクション処理

DynamoDBはトランザクションをサポートしている。注文時の在庫引き当てと注文レコード作成を原子的に行う例:

# 注文時の在庫引き当て（トランザクション）
def place_order(order_id:, product_id:, quantity:, current_stock:)
  @client.transact_write_items(
    transact_items: [
      # 在庫を減らす（在庫が足りる場合のみ）
      {
        update: {
          table_name: TABLE_NAME,
          key: { 'PK' => "PRODUCT##{product_id}", 'SK' => 'PRODUCT' },
          update_expression: 'SET stock = stock - :qty',
          condition_expression: 'stock >= :qty',
          expression_attribute_values: { ':qty' => quantity }
        }
      },
      # 注文レコードを作成
      {
        put: {
          table_name: TABLE_NAME,
          item: {
            'PK' => "ORDER##{order_id}",
            'SK' => "PRODUCT##{product_id}",
            'quantity' => quantity,
            'status' => 'confirmed',
            'created_at' => Time.now.iso8601
          }
        }
      }
    ]
  )
rescue Aws::DynamoDB::Errors::TransactionCanceledException => e
  raise InsufficientStockError, "Insufficient stock for product #{product_id}"
end

Aurora Serverless v2 — 注文データの管理

注文データはJOINが必要で集計クエリも複雑なため、Aurora Serverless v2 + PostgreSQLを選択した。

# Aurora Serverless v2 の定義
OrdersDB:
  Type: AWS::RDS::DBCluster
  Properties:
    Engine: aurora-postgresql
    EngineVersion: '15.3'
    DatabaseName: orders_db
    MasterUsername: !Sub '{{resolve:secretsmanager:${DBSecret}:SecretString:username}}'
    MasterUserPassword: !Sub '{{resolve:secretsmanager:${DBSecret}:SecretString:password}}'
    ServerlessV2ScalingConfiguration:
      MinCapacity: 0.5   # 最小 0.5 ACU (最小コスト)
      MaxCapacity: 16    # 最大 16 ACU (ピーク対応)
    EnableHttpEndpoint: true  # Data API を有効化（VPCなしでアクセス可能）
    VpcSecurityGroupIds:
      - !Ref DBSecurityGroup
    DBSubnetGroupName: !Ref DBSubnetGroup

Aurora Serverless の Data API を使うと、VPC内のLambdaを用意せずにHTTPSでDBにアクセスできる。

# Data API を使った Aurora Serverless へのアクセス
require 'aws-sdk-rdsdataservice'
 
class OrderRepository
  def initialize
    @client = Aws::RDSDataService::Client.new
    @db_arn = ENV['DB_CLUSTER_ARN']
    @secret_arn = ENV['DB_SECRET_ARN']
    @database = 'orders_db'
  end
 
  def create_order(user_id:, items:, total_price:)
    # トランザクション開始
    tx = @client.begin_transaction(
      resource_arn: @db_arn,
      secret_arn: @secret_arn,
      database: @database
    )
 
    begin
      order_id = SecureRandom.uuid
 
      # 注文レコード挿入
      execute_statement(
        sql: 'INSERT INTO orders (id, user_id, total_price, status, created_at) VALUES (:id, :user_id, :total, :status, NOW())',
        parameters: [
          { name: 'id', value: { string_value: order_id } },
          { name: 'user_id', value: { long_value: user_id } },
          { name: 'total', value: { double_value: total_price } },
          { name: 'status', value: { string_value: 'pending' } }
        ],
        transaction_id: tx.transaction_id
      )
 
      # 注文明細挿入
      items.each do |item|
        execute_statement(
          sql: 'INSERT INTO order_items (order_id, product_id, quantity, price) VALUES (:order_id, :product_id, :qty, :price)',
          parameters: [
            { name: 'order_id', value: { string_value: order_id } },
            { name: 'product_id', value: { string_value: item[:product_id] } },
            { name: 'qty', value: { long_value: item[:quantity] } },
            { name: 'price', value: { double_value: item[:price] } }
          ],
          transaction_id: tx.transaction_id
        )
      end
 
      # コミット
      @client.commit_transaction(
        resource_arn: @db_arn,
        secret_arn: @secret_arn,
        transaction_id: tx.transaction_id
      )
 
      order_id
    rescue StandardError => e
      # ロールバック
      @client.rollback_transaction(
        resource_arn: @db_arn,
        secret_arn: @secret_arn,
        transaction_id: tx.transaction_id
      )
      raise
    end
  end
 
  # 売上集計（SQLが使えるからこそできる複雑なクエリ）
  def daily_sales_summary(date:)
    result = execute_statement(
      sql: <<~SQL
        SELECT
          p.category,
          COUNT(DISTINCT o.id) as order_count,
          SUM(oi.quantity) as total_quantity,
          SUM(oi.price * oi.quantity) as total_revenue
        FROM orders o
        JOIN order_items oi ON o.id = oi.order_id
        JOIN products p ON oi.product_id = p.id
        WHERE DATE(o.created_at) = :date
          AND o.status = 'completed'
        GROUP BY p.category
        ORDER BY total_revenue DESC
      SQL
      parameters: [
        { name: 'date', value: { string_value: date.to_s } }
      ]
    )
 
    result.records.map do |row|
      {
        category: row[0].string_value,
        order_count: row[1].long_value,
        total_quantity: row[2].long_value,
        total_revenue: row[3].double_value
      }
    end
  end
 
  private
 
  def execute_statement(sql:, parameters: [], transaction_id: nil)
    params = {
      resource_arn: @db_arn,
      secret_arn: @secret_arn,
      database: @database,
      sql:,
      parameters:
    }
    params[:transaction_id] = transaction_id if transaction_id
 
    @client.execute_statement(params)
  end
end

INFO

Aurora Serverless v2 の Data API は Lambdaからの接続数問題（コネクションプールの枯渇）を解決する。通常のRDS接続ではLambdaのスケールに伴い接続数が爆発するが、Data APIはHTTPS経由なのでこの問題を回避できる。ただしData APIはレイテンシが若干高い。

DynamoDB vs Aurora の使い分けまとめ

ダイチはチーム向けにドキュメントをまとめた。

Loading diagram...

ECサイトでの使い分け:

データ選択理由
商品カタログDynamoDBカテゴリ/価格での高速検索、大量トラフィック
在庫情報DynamoDBミリ秒レベルの更新、楽観的ロック
注文データAurora Serverless複雑な集計SQL、トランザクション
ユーザー情報Aurora Serverlessリレーション（注文、レビュー等）
セッションElastiCache低レイテンシ読み書き

データストアの設計が固まり、ダイチのサーバーレスECサイトの基盤が見えてきた。次は複数の処理を組み合わせる「ワークフロー」の実装だ。

前の章イベント駆動パターン — S3・SQS・EventBridge トリガー
次の章Step Functions — ワークフローを設計する