書き込みスケーリング — シャーディングとパーティショニング

書き込みのボトルネック

読み取りのスケーリングは比較的簡単だ。レプリカを増やせばいい。しかし書き込みは違う。全ての書き込みはプライマリDBに集中するからだ。

BuzzがTVの特集番組に取り上げられた翌日、アキラはモニタリングを見て青ざめた。

Aurora Primary（翌朝の状況）:
  CPU: 87%（通常: 30%）
  書き込みIOPS: 18,500/s（上限: 20,000）
  レプリケーションラグ: 3,500ms（通常: 50ms）
  コネクション数: 95/100

エラーログ:
  [ERROR] ActiveRecord::LockWaitTimeout: Lock wait timeout exceeded
  [ERROR] PG::DeadlockDetected: ERROR: deadlock detected
  [ERROR] PG::QueryCanceled: ERROR: canceling statement due to conflict with recovery
  書き込みエラー率: 3.2%（目標: 0.1%未満）

「投稿が一気に増えてプライマリが限界に近い」とユイが言った。「いいねも、フォローも、コメントも、全部同じDBに書き込んでいるのが問題だ」

Loading diagram...

INFO

RDBMSの書き込みスケーリングは難問だ。Read Replicaは「読み取り」を分散できるが、「書き込み」は全てPrimaryに集中する。書き込みをスケールさせるには、DBを分割する必要がある。

垂直パーティショニング（機能別DB分割）

最初のアプローチはテーブルを機能ごとに別DBへ移すことだ。「Buzzの中にある複数のサービスを、別々のDBに分散させる」という考え方だ。

Loading diagram...

# config/database.yml（垂直パーティショニング後）
production:
  # ユーザー関連（書き込み少、読み取り多）
  users_primary:
    adapter: postgresql
    url: <%= ENV['USERS_DB_PRIMARY_URL'] %>
    pool: 10
 
  users_replica:
    adapter: postgresql
    url: <%= ENV['USERS_DB_REPLICA_URL'] %>
    pool: 10
    replica: true
 
  # 投稿関連（書き込み多、最も高負荷）
  posts_primary:
    adapter: postgresql
    url: <%= ENV['POSTS_DB_PRIMARY_URL'] %>
    pool: 15
 
  posts_replica:
    adapter: postgresql
    url: <%= ENV['POSTS_DB_REPLICA_URL'] %>
    pool: 15
    replica: true
 
  # ソーシャルグラフ（フォロー・いいね、書き込み多）
  social_primary:
    adapter: postgresql
    url: <%= ENV['SOCIAL_DB_PRIMARY_URL'] %>
    pool: 10

# app/models/application_record.rb
class ApplicationRecord < ActiveRecord::Base
  primary_abstract_class
end
 
# app/models/user.rb
class User < ApplicationRecord
  connects_to database: {
    writing: :users_primary,
    reading: :users_replica
  }
end
 
# app/models/post.rb
class Post < ApplicationRecord
  connects_to database: {
    writing: :posts_primary,
    reading: :posts_replica
  }
end
 
# app/models/follow.rb
class Follow < ApplicationRecord
  connects_to database: {
    writing: :social_primary,
    reading: :social_replica
  }
end
 
# app/models/like.rb
class Like < ApplicationRecord
  connects_to database: {
    writing: :social_primary,
    reading: :social_replica
  }
end

WARNING

垂直パーティショニングではクロスDBのJOINが不可能になる。User.joins(:posts) のようなクエリはエラーになる。アプリケーション層でデータを組み合わせる必要があり、これは「N+1クエリが意図的になる」ことを意味する。

# クロスDB JOIN の代替: アプリケーション層での結合
class TimelineService
  def self.build_timeline(user_id:, page: 1, per: 20)
    # 1. フォロー中のユーザーIDを取得（Social DB）
    following_ids = Follow.where(follower_id: user_id)
                          .select(:followee_id)
                          .map(&:followee_id)
 
    return [] if following_ids.empty?
 
    # 2. 投稿を取得（Posts DB）
    posts = Post.where(user_id: following_ids)
                .order(created_at: :desc)
                .offset((page - 1) * per)
                .limit(per)
                .to_a
 
    # 3. ユーザー情報を取得（Users DB）
    # N+1を避けるため、必要なIDをまとめてIN句で取得
    user_ids = posts.map(&:user_id).uniq
    users_by_id = User.where(id: user_ids)
                      .select(:id, :username, :avatar_url)
                      .index_by(&:id)
 
    # 4. アプリケーション層で結合
    posts.map do |post|
      {
        post:   post,
        author: users_by_id[post.user_id]
      }
    end
  end
end

垂直パーティショニングの効果

垂直パーティショニング後:
  各DBの書き込み負荷:
    users_db:        CPU 8%（ユーザー登録・更新が少ない）
    posts_db:        CPU 42%（投稿が多い）
    social_db:       CPU 35%（いいね・フォローが多い）
    notifications_db: CPU 15%

  合計書き込みキャパシティ: 80,000 IOPS（1つの場合の4倍）
  書き込みエラー率: 3.2% → 0.3%

水平シャーディング

垂直パーティショニングだけでは限界がある。投稿テーブルが1億行を超えたとき、次の手は水平シャーディングだ——同じ構造のDBを複数に分けてデータを分散させる。

Loading diagram...

# config/database.yml（シャーディング後）
production:
  posts_shard_0_primary:
    adapter: postgresql
    url: <%= ENV['POSTS_SHARD_0_URL'] %>
    pool: 10
  posts_shard_0_replica:
    adapter: postgresql
    url: <%= ENV['POSTS_SHARD_0_REPLICA_URL'] %>
    pool: 10
    replica: true
 
  posts_shard_1_primary:
    adapter: postgresql
    url: <%= ENV['POSTS_SHARD_1_URL'] %>
    pool: 10
  posts_shard_1_replica:
    adapter: postgresql
    url: <%= ENV['POSTS_SHARD_1_REPLICA_URL'] %>
    pool: 10
    replica: true
 
  # ... shard 2, 3 も同様

# app/models/post.rb（シャーディング対応版）
class Post < ApplicationRecord
  NUM_SHARDS = 4
 
  connects_to shards: {
    shard_0: { writing: :posts_shard_0_primary, reading: :posts_shard_0_replica },
    shard_1: { writing: :posts_shard_1_primary, reading: :posts_shard_1_replica },
    shard_2: { writing: :posts_shard_2_primary, reading: :posts_shard_2_replica },
    shard_3: { writing: :posts_shard_3_primary, reading: :posts_shard_3_replica },
  }
 
  # シャードを決定するヘルパー
  def self.shard_for(user_id)
    :"shard_#{user_id.to_i % NUM_SHARDS}"
  end
 
  # 書き込み: 必ずシャードを指定
  def self.create_for_user!(user_id:, content:, **attrs)
    connected_to(shard: shard_for(user_id), role: :writing) do
      create!(user_id: user_id, content: content, **attrs)
    end
  end
 
  # 読み取り: ユーザーIDでシャードを決定
  def self.posts_by_user(user_id:, page: 1, per: 20)
    connected_to(shard: shard_for(user_id), role: :reading) do
      where(user_id: user_id)
        .order(created_at: :desc)
        .offset((page - 1) * per)
        .limit(per)
        .to_a
    end
  end
end

タイムラインのファンアウト問題

シャーディング後、タイムライン取得が複雑になる。フォロー中のユーザーが複数のシャードに散らばるからだ。

# app/services/timeline_service.rb
class TimelineService
  def self.get_timeline(user_id:, limit: 50)
    following_ids = Follow.where(follower_id: user_id).pluck(:followee_id)
    return [] if following_ids.empty?
 
    # フォロー中のユーザーをシャードごとにグループ化
    shard_groups = following_ids.group_by { |uid| Post.shard_for(uid) }
 
    # 各シャードから並列でフェッチ（concurrent-ruby）
    futures = shard_groups.map do |shard, user_ids|
      Concurrent::Future.execute do
        Post.connected_to(shard: shard, role: :reading) do
          Post.where(user_id: user_ids)
              .where(created_at: 7.days.ago..)
              .order(created_at: :desc)
              .limit(limit)
              .select(:id, :user_id, :content, :likes_count, :created_at)
              .to_a
        end
      end
    end
 
    # タイムアウト付きで結果を収集
    all_posts = futures.flat_map do |future|
      future.value(5)  # 5秒でタイムアウト
    rescue Concurrent::TimeoutError
      Rails.logger.error "Timeline shard timeout"
      []
    end
 
    # 全シャードの結果をマージしてソート
    all_posts.sort_by(&:created_at).reverse.first(limit)
  end
end

INFO

ファンアウト vs ファンイン: Twitterのような大規模SNSでは、タイムライン生成を「投稿時にフォロワーのキャッシュに書き込む（ファンアウト書き込み）」方式にすることが多い。Buzzも50万フォロワーを超えるインフルエンサーが増えてきたら、このアプローチを検討する必要がある。フォロワーが多い人（有名人）は「ファンイン（読み取り時にマージ）」で処理するハイブリッド方式が実用的だ。

シャーディングキーの選び方

シャーディングで最も重要なのはシャーディングキーだ。一度決めたら変更が非常に困難なため、慎重に選ぶ。

キー候補	メリット	デメリット	適した用途
`user_id`	ユーザーごとのデータが集まる	ホットスポット（有名人）	SNS投稿
`post_id`	均等分散	タイムライン取得が複雑	コメント
`created_at`	時系列クエリが速い	新しいシャードに集中	ログ
ハッシュ値	均等分散	範囲クエリが困難	KV
SnowflakeID	時系列＋均等分散	専用ライブラリ必要	汎用

BuzzではSnowflakeIDを採用した。時系列でソートできながら、シャードも均等に分散できる。

# lib/buzz/snowflake_id.rb
module Buzz
  class SnowflakeId
    # Twitter の Snowflake アルゴリズムに基づく
    # 64ビット = 41bit(タイムスタンプ) + 10bit(マシンID) + 12bit(シーケンス)
 
    EPOCH = Time.utc(2024, 1, 1).to_i * 1000  # Buzz サービス開始のエポック（ミリ秒）
 
    # 環境変数でマシンIDを設定（ECSタスクごとに異なるIDを付与）
    MACHINE_ID = ENV.fetch('MACHINE_ID', SecureRandom.random_number(1024)).to_i & 0x3FF
 
    def self.generate
      timestamp = (Time.current.to_f * 1000).to_i - EPOCH
      machine   = MACHINE_ID & 0x3FF      # 10ビット（1024台まで）
      sequence  = SecureRandom.random_number(4096) & 0xFFF  # 12ビット
 
      (timestamp << 22) | (machine << 12) | sequence
    end
 
    def self.shard_for(id)
      # IDの下位ビットはランダム → 均等分散
      id.to_i % Post::NUM_SHARDS
    end
 
    def self.created_at_from(id)
      # IDからタイムスタンプを復元（デバッグや問い合わせ対応に便利）
      timestamp_ms = (id.to_i >> 22) + EPOCH
      Time.at(timestamp_ms / 1000.0)
    end
  end
end
 
# 使用例
id = Buzz::SnowflakeId.generate
# => 1234567890123456789
 
Buzz::SnowflakeId.created_at_from(id)
# => 2024-06-15 14:30:00 +0900
 
Buzz::SnowflakeId.shard_for(id)
# => 2  （シャード2に書き込む）

# app/models/post.rb にSnowflakeIDを組み込む
class Post < ApplicationRecord
  before_create :assign_snowflake_id
 
  private
 
  def assign_snowflake_id
    self.id ||= Buzz::SnowflakeId.generate
  end
end

楽観的ロックで競合を減らす

書き込みの競合を減らすには楽観的ロックが効果的だ。「書き込む時に競合を検出する」アプローチで、悲観的ロック（先にロックを取る）よりスループットが高い。

# マイグレーション: lock_version カラムを追加
class AddLockVersionToPosts < ActiveRecord::Migration[7.1]
  def change
    add_column :posts, :lock_version, :integer, default: 0, null: false
    # lock_version カラムがあれば、Railsが自動で楽観的ロックを有効にする
  end
end
 
# 使い方
post1 = Post.find(123)  # lock_version: 5
post2 = Post.find(123)  # lock_version: 5（同じ）
 
# post1 が先に更新
post1.update!(content: "最初の更新")  # lock_version → 6
 
# post2 が後から更新しようとすると...
begin
  post2.update!(content: "後からの更新")  # lock_version: 5 ≠ DBの 6
rescue ActiveRecord::StaleObjectError => e
  # 競合を検出！ユーザーに通知
  render json: {
    error: 'conflict',
    message: '他のユーザーが同じ投稿を更新しました。最新版を確認してください。',
    current_version: Post.find(123).content
  }, status: :conflict
end

# より安全ないいね処理（楽観的ロックを使わない場合）
class Like < ApplicationRecord
  # いいね数の更新に楽観的ロックは使わない
  # 代わりにAtomicなSQLでカウントを更新する
 
  after_create  :increment_post_likes_count
  after_destroy :decrement_post_likes_count
 
  private
 
  def increment_post_likes_count
    # update! より update_all が安全（オブジェクトを介さない）
    Post.where(id: post_id).update_all('likes_count = likes_count + 1')
  end
 
  def decrement_post_likes_count
    Post.where(id: post_id).update_all('likes_count = GREATEST(likes_count - 1, 0)')
  end
end

バッチ書き込みで書き込み回数を削減

個別の書き込みを束ねることで、DBへの往復回数を減らす。

# 悪い例: 1行ずつ INSERT（N回のDB往復）
follower_ids.each do |follower_id|
  Notification.create!(
    recipient_id: follower_id,
    type: 'new_post',
    post_id: post.id
  )
end
 
# 良い例: 一括 INSERT（1回のDB往復）
notifications = follower_ids.map do |follower_id|
  {
    recipient_id: follower_id,
    type: 'new_post',
    post_id: post.id,
    created_at: Time.current,
    updated_at: Time.current
  }
end
 
# 1回のSQLで全件挿入（Rails 6以降）
Notification.insert_all(notifications, returning: [:id])
# INSERT INTO notifications (...) VALUES (...), (...), (...) ...
 
# 重複を無視したい場合
Notification.insert_all(notifications,
  unique_by: [:recipient_id, :post_id])
 
# UPSERT（挿入 or 更新）
Notification.upsert_all(notifications,
  unique_by: [:recipient_id, :post_id],
  update_only: [:updated_at])

# バルク更新（複数行を一度に更新）
# Gemfile: gem 'activerecord-import'
 
posts_to_update = Post.where(user_id: banned_user_id).to_a
posts_to_update.each { |p| p.status = 'hidden' }
 
# 一括更新
Post.import posts_to_update,
  on_duplicate_key_update: { conflict_target: [:id], columns: [:status] }

書き込みキューでバースト吸収

突発的な書き込みバーストを吸収するためにキューを挟む。

# app/services/like_service.rb
class LikeService
  REDIS = Redis.new(url: ENV['REDIS_URL'])
  BATCH_SIZE = 100
  FLUSH_INTERVAL = 5.seconds
 
  # いいねを Redis に蓄積（瞬時に返る）
  def self.record_like(post_id:, user_id:)
    event = { post_id: post_id, user_id: user_id, ts: Time.current.to_i }
    REDIS.rpush('likes:pending', event.to_json)
 
    # バッファが一定数を超えたら非同期でフラッシュ
    count = REDIS.llen('likes:pending')
    FlushLikesJob.perform_later if count >= BATCH_SIZE
  end
 
  # バッチでDBに書き込む
  def self.flush_pending_likes
    likes_json = REDIS.lrange('likes:pending', 0, BATCH_SIZE - 1)
    return if likes_json.empty?
 
    likes = likes_json.map { |j| JSON.parse(j, symbolize_names: true) }
 
    # DB に一括書き込み
    Like.insert_all(
      likes.map { |l|
        { post_id: l[:post_id], user_id: l[:user_id], created_at: Time.at(l[:ts]) }
      },
      unique_by: [:post_id, :user_id]
    )
 
    # Post の likes_count を一括更新
    post_ids = likes.map { |l| l[:post_id] }.tally
    post_ids.each do |post_id, count|
      Post.where(id: post_id).update_all("likes_count = likes_count + #{count}")
    end
 
    # 処理済みを Redis から削除
    REDIS.ltrim('likes:pending', likes_json.count, -1)
  end
end

改善後の書き込みパフォーマンス

Loading diagram...

指標	シャーディング前	シャーディング後
書き込みIOPS上限	20,000	80,000（4倍）
書き込みエラー率	3.2%	0.02%
デッドロック発生率	高（日次数十件）	低（月次数件）
最大書き込みスループット	500 RPS	2,000 RPS
月次コスト	$340（1台）	$1,200（4台）

「書き込みも4倍にスケールした」アキラは言った。「でもアプリサーバー自体は？リクエストが増えればアプリも増やさないといけない」

次章では、ECS Auto ScalingとALBでRailsアプリ自体を水平展開する方法を学ぶ。

付録: シャーディングの落とし穴

よくある失敗:

1. シャーディングキーを後から変更しようとする
   → データ移行が地獄。最初から慎重に選ぶ

2. クロスシャードJOINを使おうとする
   → 不可能。アプリ層でのデータ結合か、
     読み取り専用の非正規化テーブルを用意する

3. シャード数が少なすぎる → 後で増やすと再シャーディングが必要
   シャード数が多すぎる → 管理コストが増大
   現実的な出発点: 4〜8シャード

4. ホットスポット
   有名人の user_id が特定のシャードに集中
   → ConsistentHashingで対策（シャード変更時のデータ移動を最小化）

5. スキーマ変更が4倍の手間になる
   → マイグレーションを各シャードに適用する仕組みが必要

# 全シャードにマイグレーションを適用するRakeタスク
namespace :db do
  namespace :shards do
    desc "全シャードにマイグレーションを適用"
    task migrate: :environment do
      Post::NUM_SHARDS.times do |i|
        shard = :"shard_#{i}"
        puts "Migrating shard #{shard}..."
        Post.connected_to(shard: shard, role: :writing) do
          ActiveRecord::MigrationContext.new(
            Rails.root.join('db', 'migrate').to_s,
            ActiveRecord::SchemaMigration
          ).migrate
        end
      end
      puts "All shards migrated successfully"
    end
  end
end