技術カタログ | Developer's Catalog

データベース

Riak KV

概要

Riak KVは、高可用性、耐障害性、および運用の簡素性を重視して設計された分散型NoSQLキー・バリューデータベースです。Amazon DynamoペーパーとCAP定理の原則に強く影響を受け、Erlangで実装されています。マスターレスアーキテクチャを採用し、すべてのノードが読み書きリクエストに対応可能で、クラスタ全体でデータを自動分散・レプリケーションします。

詳細

Riak KVは2009年にBasho Technologiesによって開発され、現在はRiak社が開発・サポートを継続しています。分散システムの「Ring」アーキテクチャを採用し、160ビットのキー空間でデータを論理的に分散配置します。各ノードは複数のVirtual Nodes（vnodes）を持ち、データの断片を担当します。

Riak KVの主な特徴：

マスターレスアーキテクチャ（全ノードが平等）
自動データ分散とレプリケーション
Vector Clocksによる因果関係追跡
調整可能な一貫性レベル（R/W値）
複数のストレージバックエンド対応
Gossipプロトコルによるクラスタ状態共有
HTTP APIとProtocol Buffers API
ネットワーク分断耐性
水平スケーラビリティ
結果整合性（Eventually Consistent）

ストレージバックエンドとしてBitcask（デフォルト）、LevelDB、Memory、Leveledなどを選択できます。

メリット・デメリット

メリット

高可用性: 24/7運用環境に最適、複数ノード障害でも動作継続
水平スケーラビリティ: ノード追加による線形的な性能向上
運用簡易性: マスターレス設計で管理が容易、シングルポイントオブフェイラー無し
耐障害性: 自動フェイルオーバーとデータ復旧
柔軟な一貫性: R/W値による一貫性レベル調整
マルチデータセンター対応: 地理的に分散した環境をサポート
結果整合性: CAP定理のA（可用性）とP（分断耐性）を優先

デメリット

複雑なクエリ不可: SQL式のSELECT文やJOINは不可能
小規模環境には過剰: 最低5ノード構成が推奨されコストが高い
強一貫性不適: ACID特性が必要な用途には不向き
大きなオブジェクト不向き: 1MB以下のデータに最適化
学習コスト: 分散システムの概念理解が必要
開発の複雑化: 競合解決をアプリケーション側で実装

主要リンク

書き方の例

インストール・セットアップ

# Ubuntu/Debian
curl -s https://packagecloud.io/install/repositories/basho/riak/script.deb.sh | sudo bash
sudo apt-get install riak

# Red Hat/CentOS
curl -s https://packagecloud.io/install/repositories/basho/riak/script.rpm.sh | sudo bash
sudo yum install riak

# macOS (Homebrew)
brew tap basho/riak
brew install riak

# Docker
docker run -d --name riak-container -p 8087:8087 -p 8098:8098 basho/riak-kv

# サービス開始
sudo systemctl start riak
sudo systemctl enable riak

# 動作確認
curl http://127.0.0.1:8098/ping
# Expected: OK

基本操作（HTTP API）

# データ保存
curl -X PUT -H "content-type: text/plain" \
  http://127.0.0.1:8098/riak/users/user1 \
  --data '{"name": "田中太郎", "age": 30, "email": "[email protected]"}'

# データ取得
curl -i http://127.0.0.1:8098/riak/users/user1

# 条件付き取得（一貫性レベル指定）
curl http://127.0.0.1:8098/riak/users/user1?r=2

# データ更新
curl -X PUT -H "content-type: application/json" \
  -H "X-Riak-Vclock: a85hYGBgzGDKBVIcypz/fgaUHjmdwZTImMfKyDt97Q==" \
  http://127.0.0.1:8098/riak/users/user1 \
  --data '{"name": "田中太郎", "age": 31, "email": "[email protected]"}'

# データ削除
curl -X DELETE http://127.0.0.1:8098/riak/users/user1

# キー一覧（開発用のみ）
curl http://127.0.0.1:8098/riak/users?keys=true

バケット設定

# バケット設定取得
curl http://127.0.0.1:8098/riak/users

# レプリケーション数設定（N値）
curl -X PUT -H "content-type: application/json" \
  http://127.0.0.1:8098/riak/users \
  --data '{"props":{"n_val":5}}'

# 一貫性レベル設定
curl -X PUT -H "content-type: application/json" \
  http://127.0.0.1:8098/riak/users \
  --data '{"props":{"r":"quorum","w":"quorum","dw":"quorum"}}'

# 競合解決設定
curl -X PUT -H "content-type: application/json" \
  http://127.0.0.1:8098/riak/users \
  --data '{"props":{"allow_mult":true,"last_write_wins":false}}'

Node.js クライアント使用例

const RiakClient = require('basho-riak-client')

// クライアント作成
const client = new RiakClient(['127.0.0.1:8087'])

// データ保存
const storeValue = {
    bucket: 'users',
    key: 'user1',
    value: JSON.stringify({
        name: '田中太郎',
        age: 30,
        email: '[email protected]'
    }),
    contentType: 'application/json'
}

client.storeValue(storeValue, (err, rslt) => {
    if (err) {
        console.error('保存エラー:', err)
    } else {
        console.log('データ保存成功')
        console.log('Vector Clock:', rslt.vclock)
    }
})

// データ取得
const fetchValue = {
    bucket: 'users',
    key: 'user1'
}

client.fetchValue(fetchValue, (err, rslt) => {
    if (err) {
        console.error('取得エラー:', err)
    } else if (rslt.values.length === 0) {
        console.log('データが見つかりません')
    } else {
        const userData = JSON.parse(rslt.values[0].value.toString())
        console.log('ユーザーデータ:', userData)
    }
})

// 複数バージョン処理（競合解決）
client.fetchValue(fetchValue, (err, rslt) => {
    if (rslt.values.length > 1) {
        console.log('競合が検出されました')
        rslt.values.forEach((value, index) => {
            console.log(`バージョン ${index + 1}:`, 
                JSON.parse(value.value.toString()))
        })
        
        // 最新の更新時刻のデータを選択（例）
        const latest = rslt.values.reduce((prev, current) => {
            const prevData = JSON.parse(prev.value.toString())
            const currentData = JSON.parse(current.value.toString())
            return currentData.updatedAt > prevData.updatedAt ? current : prev
        })
        
        console.log('選択されたバージョン:', JSON.parse(latest.value.toString()))
    }
})

// 接続終了
client.stop()

クラスタ管理

# クラスタ状態確認
riak-admin cluster status

# 新ノード追加
riak-admin cluster join [email protected]

# クラスタプラン表示
riak-admin cluster plan

# プラン実行
riak-admin cluster commit

# ノード退去
riak-admin cluster leave [email protected]

# リング状態確認
riak-admin ring-status

# 手動でハンドオフ実行
riak-admin transfer-limit 4
riak-admin handoff status

監視・保守

# ノード統計情報
riak-admin stat

# 特定統計のみ表示
riak-admin stat | grep -E "(ring_|handoff_|coord_redirs)"

# vnodeの状態確認
riak-admin vnode-status

# パフォーマンス診断
riak-admin diag

# AAE（Anti-Entropy）状態
riak-admin aae-status

# 強制修復実行
riak-admin repair-2i users

# バックアップ（データディレクトリ）
sudo tar -czf riak-backup-$(date +%Y%m%d).tar.gz /var/lib/riak

設定ファイル例（riak.conf）

# ノード設定
nodename = [email protected]
distributed_cookie = riak

# ディレクトリ設定
platform_data_dir = /var/lib/riak
platform_etc_dir = /etc/riak
platform_log_dir = /var/log/riak

# ネットワーク設定
listener.http.internal = 127.0.0.1:8098
listener.protobuf.internal = 127.0.0.1:8087

# ストレージ設定
storage_backend = bitcask

# クラスタ設定
ring_size = 64

# デフォルトバケット設定
buckets.default.n_val = 3
buckets.default.r = 2
buckets.default.w = 2
buckets.default.dw = 1

# パフォーマンス設定
erlang.schedulers.force_wakeup_interval = 500
erlang.schedulers.compaction_of_load = false

# ログ設定
log.error.file = /var/log/riak/error.log
log.console.level = info
log.crash.file = /var/log/riak/crash.log

ベストプラクティス

# 1. 適切なクラスタサイズ（最低5ノード）
# 3ノード: 開発環境のみ
# 5ノード: 本番環境推奨
# 7ノード以上: 大規模環境

# 2. データモデリング
# - オブジェクトサイズ: 1MB以下
# - 自然キー使用: ユーザーID、タイムスタンプ等
# - バケット設計: 論理的なデータグルーピング

# 3. 一貫性レベル設定
# R + W > N: 強一貫性
# R = 1, W = 1: 高可用性優先
# R = quorum, W = quorum: バランス型

# 4. 運用時の注意点
# - 同時ノード障害は N/2 未満に留める
# - 定期的なAAE実行で整合性確認
# - ハンドオフ完了を待ってから次のノード操作
# - 監視項目: CPU 30%以下、I/O 90%以下

# 5. パフォーマンス最適化
# - Bitcaskの定期的なマージ実行
# - 適切なerlang.schedulers設定
# - ネットワーク帯域幅の確保
# - SSDストレージの使用推奨