原文: Cassandra - A Decentralized Structured Storage System
This article is translated by @ono_matope. Please contact me if any problem.
Cassandraはたくさんのコモディティなサーバー上に分散した非常に大量の構造化データを管理するための分散ストレージシステムで、単一障害点なしに高可用性なサービスを提供しています。Cassandraは数百ノードの（もしかしたら複数データセンターにわたる）インフラ上で動作することを目標としています。このスケールでは、大小のコンポーネントは継続的に壊れます。Cassandraがこれらの障害に直面しながら永続状態を管理する方法はこのサービスに依存するソフトウェアシステムの信頼性とスケーラビリティを供給します。 Cassandraは多くの面でデータベースに似ていて、設計や実装戦略を共有していますが、Cassandraは完全なリレーショナルデータモデルをサポートせず、かわりにデータレイアウトとフォーマットにわたる動的な制御をサポートするシンプルなデータモデルをサポートします。Cassandraシステムは安価なコモディティハードウェア上で動作し、読み込み効率を犠牲にすることなく高い書き込みスループットをハンドリングするために設計されました。
Facebookはピーク時数億ユーザーを世界中の多数のデータセンターに配置された数万のサーバーを使ってサービスする世界最大のソーシャルネットワーキングプラットフォームを運用しています。Facebookのプラットフォームにはパフォーマンス、信頼性、効率性、そして継続的成長を支えるためにプラットフォームは高度にスケーラブルである必要があります。数千のコンポーネントで構成された単一のインフラストラクチャ上で障害を扱うことは我々にとっての平常モードで、そこでは常に少数ながら重大な数のサーバーやネットワークコンポーネントがいつでも壊れ続けています。そういうものとして、ソフトウェアシステムは障害を例外ではなく標準として構築される必要があります。上で述べた信頼性とスケーラビリティのニーズを満たすためにFacebookはCassandraを開発しました。
Cassandraはスケーラビリティと可用性を達成するためによく知られた技術の統合を用います。CassandraはInbox Search問題のストレージニーズを満たすために設計されました。Inbox SearchはユーザーがFacebook Inboxを検索できるようにする機能です。Facebookにおいては、これはシステムが非常に高いwriteスループット、一日あたり数十億write、そしてまたユーザー数のスケールのハンドリングを要求されていることを意味しました。ユーザーは地理的に分散したデータセンターにサービスされるため、データセンター間でデータをレプリケート出来るようにすることは検索レイテンシを低く抑えるためのキーでした。Inbox Searchは2008年6月に約1億ユーザーを対象にリリースされ、現在は2億5千万ユーザーですが、今のところCasandraは役割を果たしてきました。CassandraはいまやFacebookの複数のサービスのバックエンドストレージシステムとしてデプロイされています。
この論文は以下のように構成されています。Section 2では関連事例を紹介しますが、そのいくつかは我々の設計に多大な影響を与えています。Section 3ではデータモデルの詳細を説明します。Section 4ではクライアントAPIの概要を説明します。Section 5ではシステムデザインとCassandraを動かす分散アルゴリズムを説明します。Section 6ではCassandraを動かす経験とパフォーマンスを向上するための改良の詳細。Section 6.1ではFacebookプラットフォームのアプリケーションの一つがどのようにCassandraを使うのか。最後にSection 7ではCassandraの将来の展望でまとめます。
パフォーマンス、可用性、そして永続性のためにデータを分散することはファイルシステムとデータベースのコミュニティにおいて広く研究されてきました。フラットな名前空間しかサポートしないP2Pストレージシステムと比べて、通常分散ファイルシステムは階層的な名前空間をサポートします。Ficus[14]やCoda[16]のようなシステムは一貫性を犠牲にして高可用性のためにファイルをレプリケートします。更新衝突は通常特別な衝突解決手続きを用いて管理されます。Farsite[2]はいかなる中央管理されたサーバーも用いない分散ファイルシステムです。Farsiteはレプリケーションを用いて高い可用性とスケーラビリティを達成しています。Google File System(GFS)[9]はGoogleの内部アプリケーションの状態をホストするために作られたもう一つの分散ファイルシステムです。GFSは全てのメタデータとをホストする単一のマスターサーバーとチャンクに分割されたデータがストアされるチャンクサーバーによるシンプルな設計です。しかし、GFSマスターはChubby[3]抽象化による障害耐性を実現しました。Bayou[13]は切断されたオペレーションを許可し、結果一貫性を提供する分散リレーショナルデータベースです。これらのシステム、Bayou, Coda, Ficusは切断されたオペレーションを許可し、ネットワーク分断や停電への弾力性を実現しました。これらのシステムは衝突解決手続きにおいて異なります。例えば、CodaとFicusはシステムレベルの衝突解決を行い、Bayouはアプリケーションレベルの解決をゆるしています。しかしそれらは全て、結果一貫性を保証しています。これらのシステムと似て、Dynamo[6]はread/writeオペレーションをネットワーク分断の最中であろうとも実行し、更新衝突を異なる衝突解決メカニズム（クライアントドリブンなものを含む）で解決します。伝統的なレプリケーションつきリレーショナルデータベースシステムはレプリケーションデータの強一貫性保証の問題に注力しています。強一貫性はアプリケーションの書き手に便利なプログラミングモデルを提供する一方、これらのシステムはスケーラビリティと可用性の面で制約を受けます[10]。これらのシステムは通常強一貫性を保証しているので、ネットワーク分断をハンドリングする能力はありません。
Dynamo[6]はAmazonがユーザーのショッピングカートをストアして取得するために使っているストレージシステムです。DynamoのGossipベースのメンバーシップアルゴリズムは全てのノードが他のノード全ての情報を得ることを助けます。Dynamoはリクエストをほぼ1ホップでルーティングする構造化オーバーレイと定義できます。Dynamoはvector clock手法を用いて更新衝突を検出しますが、クライアントサイドの衝突解決手法を好みます。Dynamoのwrite命令はvector timestampを管理するためにreadも行う必要があります。これは高いwriteスループットが必要なシステム環境においてはとても大きな制約になり得ます。Bigtable[4]は構造とデータ分散の両方を提供しますが、その永続性を分散ファイルシステムに依存しています。
Cassandraのテーブルはひとつのkeyによってインデックスされた多次元マップです。値は高度に構造化されたひとつのオブジェクトです。テーブルの行キーはサイズ制限のない文字列ですが、通常16から36バイト長です。単一の行キー下の全てのオペレーションはどれほどたくさんのカラムが読み書きされようとレプリカごとにアトミックです。CassandraはBigtable[4]システムで行われているものととてもよく似たカラムファミリと呼ばれる集合に分類されます。Cassandraは二つの種類のカラムファミリを持ち、それらはシンプルカラムファミリとスーパーカラムファミリです。スーパーカラムファミリはカラムファミリのカラムファミリとして可視化できます。
さらに、アプリケーションはスーパーカラムかシンプルカラム内のカラムのソート順を指定できます。システムはカラムを時間か名前でソートすることをゆるします。カラムの時間ソートはInbox Searchのように結果が常に時系列順に表示されるアプリケーションに利用されます。カラムファミリの中の全てのカラムはカラムファミリ：カラムという規約を使ってアクセスされ、カラムファミリの中の全てのsuper型のカラムはカラムファミリ：スーパーカラム：カラムという規約を使ってアクセスされます。スーパーカラムファミリの抽象化のパワーのいい例はSection 6.1にあります。通常アプリケーションは専用のCassandraクラスターを使用し彼らのサービスの一部として管理します。システムは複数テーブルの概念をサポートするにもかかわらず、全てのデプロイは彼らのスキーマ内に一つしかテーブルを持ちません。
Cassandra APIは次の3つのシンプルなメソッドからなります。
- insert(table,key,rowMutation)
- get(table,key,columnName)
- delete(table,key,columnName)
columnNameはカラムファミリ内の特定のカラム、カラムファミリ、スーパーカラムファミリ、あるいはスーパーカラム内のカラムを指すかもしれません。
プロダクション設定で運用される必要のあるストレージシステムのアーキテクチャは複雑です。実際のデータ永続かコンポーネントに加え、システムは次の特性を備えておく必要があります：スケーラブルで堅牢なロードバランシングソリューション、メンバーシップと障害の検知、障害復旧、レプリカ同期、過負荷対策、状態転送、並列性とジョブスケジューリング、リクエスト整列、リクエストルーティング、システムモニタリングとアラーム、そして設定管理。それぞれを詳しく書き出すことはこの論文のスコープを越えてしまうので、ここではCassandraで用いられている小穴分散システム技術：パーティショニング、レプリケーション、メンバーシップ、障害対応とスケーリングにフォーカスしましょう。これら全てのモジュールはread/writeリクエストを扱うために同時的に動作します。典型的には、あるキーへのread/writeリクエストはCassandraクラスター内のいずれかのノードにルーティングされます。その時そのノードはその特定のキーのレプリカを特定します。writeであれば、システムはリクエストをそのレプリカにルーティングし、定足数のレプリカがwriteの完了を承認するのを待ちます。readであれば、クライアントに要求された一貫性保証に基づき、システムはリクエストを最近傍のレプリカか、全てのレプリカに転送し、定足数の反応を待ちます。
Cassandraのキーとなる設計の特徴の一つは、線形なスケールの能力です。これは、クラスター内でノード(すなわち、ストレージホスト)の集合をまたいでのデータの動的パーティションを要求します。Cassandraはcosistent hashing[11]を用いてクラスター内でデータをパーティショニングしますが、それを行うのにorder preserving hash functionを使います。consistent hashingではハッシュ関数の出力範囲は固定された環状の空間、または"リング"として扱われます(ハッシュの最大値は最小値をラップします)。
システムの各ノードはリング中でその位置を表すこの空間の中でランダムな値に割り当てられます。各データアイテムはリング上のポジションを割り出すためにデータアイテムのキーをハッシングしてノードに割り当てられたキーによって識別され、そしてリングを時計回りにまわり、アイテムのポジションより大きな位置にある最初のノードを見つけます。このノードはこのキーのコーディネーターとされます。アプリケーションはこのキーを指定し、Cassandraはそれをリクエストをルーティングするために使います。従って、各ノードはそれとリング上でそれの一つ前のノードとの間の領域について責任を持つことになります。consitent hashingの主要な利点はノードの離脱と出現が直接の隣にしか影響を与えず、その他には影響がないということです。標準的なconsistent hashingアルゴリズムにはいくつかの挑戦があります。最初に、リング上の各ノードのランダムな割り当てはデータと負荷分散の不均等を招くということです。次に、標準的なアルゴリズムはノードのパフォーマンスにおける異種混交性を念慮していません。通常、この問題を解決する二つの方法が存在します：ひとつは、ノードを環の複数の位置に割り当てるということです(Dynamoが行っているように)、二つ目は、リング上の負荷情報を分析し、[17]で述べられているように、リング上の負荷の軽いノードを、負荷の高いノードを緩和するために移動させることです。Cassandraは後者を選択したので、設計と実装をとても素直になり、ロードバランシングにかんする決定論的な選択を助けました。
Cassandraは高可用性と高永続性を達成するためにレプリケーションを使います。各データアイテムはN個のホストにレプリケーションされますが、Nは"インスタンスごと"に設定されたレプリケーションファクターです。各キー, kはコーディネーターノード(前章で述べた)に割り当てられます。コーディネーターはそのレンジに落ちてきたデータアイテムのレプリケーションの責任があります。そのレンジ内の書くキーをローカルにストアするのに加え、コーディネーターはそれらのキーをリング上のN-1個のノードにレプリケーションします。Cassandraはデータがどのようにレプリケーションされなくてはならないかというたくさんのオプションをクライアントに提供します。Cassandraは"Rack Unaware", "Rack Aware"(データセンター内), "Datacenter Aware"などの多様なレプリケーションポリシーを提供します。"Rack Aware"と"Datacenter Aware"戦略の場合、アルゴリズムは多少複雑になります。
CassandraシステムはZookeeper[13]と呼ばれるシステムを用いてそのノード間でリーダーを選出します。クラスターに参加している全てのノードはそのレプリカがどのレンジをレプリケートするか教えてくれるリーダーに問い合わせ、リーダーはリング上でN-1個より多いレンジを担当するノードが出ないように協調努力します。あるノードが受け持つレンジに関するメタデータは各ノードのローカル、障害耐性的なやり方でZookeeper内にキャッシュされますーこの方法では、壊れて復帰したノードはどのレンジに責任があるのか知っています。我々はDynamoから用語を拝借し、ノードは与えられたレンジ"preference list"に対して責任があると考えます。
Section 5.1で説明したように、全てのノードはシステム内の他の全てのノードを知っているので、従って彼らが責任を持っている範囲も知っています。CassandraはSection 5.2で述べたようなリラックスした定足数要求によってノード障害とネットワーク分断の状況下で永続性保証を提供します。データセンター障害は停電、空調の故障、ネットワーク障害、自然災害などにより引き起こされます。Cassandraは各行が複数データセンターにレプリケーションされるように設定されます。本質的に、キーのpreference listは複数のデータセンターに散らばったストレージノードから構成されています。データセンター間は高速ネットワークリンクで接続されています。この複数データセンターにわたるレプリケーション手法は完全なデータセンター障害に無停止で対応することが出来ます。
Cassandraのクラスターメンバーシップはとても効率的な反エントリピーGossipベース機構のScuttlebutt[19]に基づいています。Scuttlebuttの顕著な特徴は、それがCPUとgossipチャンネルをとても効率的に利用することです。Cassandraシステム内でGossipはメンバーシップだけでなく、他の関連したコントロール状態を散布するためにも使われます。
障害検知はノードがローカルで他のノードが生きているかどうかを判断出来る機構です。Cassandraでは障害検知は様々な命令の間に到達不能なノードと無駄な通信を試みるのを避けるためにも使われます。CassandraはΦ Accrual Failure Detector[8]の修正バージョンを用います。アクルーアル障害検知のアイデアは、障害検知モジュールはノードが生きているか死んでいるかという状態を表すBoolean値は発行しないというものです。障害検知のかわりに、モジュールは各モニータ対象のノードの容疑レベルを表す値を発行します。この値はΦと定義されます。基本的な考えは、Φの値をネットワークとモニター対象のノードの負荷を反映して動的に補正されるスケール上で表現することです。Φは次の意味を持ちます：いくつかのしきい値Φが与えられたとき、Φ=1のときノードAに容疑を決定すると、そのとき我々が間違っているという尤度(すなわち、その決定が未来においてきっぱりと否定される確からしさ)は約10%です。その尤度はΦ=2のとき1%、Φ=3のとき0.1%…です。システムの全てのノードはクラスタ内の他のノードからのgossipメッセージの到達時間の平滑化ウィンドウを保持します。これらの到達時間の分散が決定され、Φが計算されます。オリジナルの論文は分散はガウス分布に近似すると暗示しているにも関わらず、gossipチャンネルの性質とそのレイテンシへのインパクトから、指数分布がよりよい近似だと気づきました。我々の知識のため、我々のGossipベースな設定のAccrual Failure Detectionの実装はその最初の種類のものです。Accrual Failure Detectorsは正確さとスピードの両面で良く、またネットワーク状態とサーバーフか状態にもよく補正します。
ノードが最初にスタートする時、リング中の自分の位置のためにランダムなトークンを選択します。障害耐性のため、マッピングはローカルディスクのほか、Zookeeperにも永続化されます。トークン情報はそのときcluster中にgossipされます。これは我々が全ノードとそれらのリング中でのめいめいの位置を知る方法です。これはどのノードでもキーへのリクエストをクラスタ中の正しいノードにルーティングすることを可能にします。bootstrapの場合、ノードがclusterに参加する必要があるとき、それはいくつかのクラスタ内のコンタクトポイントを列挙した設定ファイルを読み込みます。これらを我々を初期コンタクトポイント、クラスタの種、と呼びます。種はZookeeperのようなコンフィグサービスからももたらされます。
Facebookの環境では、ノード停止(障害やメンテナンスタスクによる)はしばしば一時的ですが、間隔を超過するかもしれません。障害はディスク障害、CPU不良その他など、様々な形態を持ちえます。ひとつのノード停止が永続的な離脱の兆候であることは稀なので、パーティション割り当てのリバランシングや到達不能レプリカの修復に帰結するべきではありません。同じように、人的エラーはCassandraノードの意図しないスタートアップを導くかもしれません。このようなことから、全てのメッセージは各Cassandraインスタンスのクラスター名を含んでいます。もし設定エラーがノードを誤ったCassandraインスタンスにジョインさせようとした場合、クラスター名に基づいて中断されます。これらの理由から、Cassandraインスタンスへのノードの追加と削除は、明示的なメカニズムで開始するのが適切とされています。管理者はコマンドラインツールかブラウザーでCassandraノードに接続し、クラスターへのjoinかleaveのメンバーシップ変更を発行します。
あるノードがシステムに追加された時、負荷の高いノードを緩和できるトークンが割り当てられます。これは、以前は他のノードが責任を持っていたレンジを新しいノードが分割するということです。Cassandraブートストラップアルゴリズムは、コマンドラインユーティリティかCassandra webダッシュボードを使うオペレーターによって、システム内の他のどのノードからも開始されます。
データを諦めたノードはカーネルコピー技法を使って新しいノードにストリームします。運用経験上、データは単一ノードから40MB/secのレートで転送できます。我々は、複数レプリカをブートストラップ転送に参加させて、Bittorrentのように並列化することで改善することに取り組んでいます。
Cassandraシステムはデータ永続化に関して、ローカルのファイルシステムに依存しています。データは効率的なデータ取得に適したフォーマットを用いてディスク上に表現されます。典型的なwrite命令は永続性とリカバリビリティのためcommit logに書き込み、インメモリのデータ構造を更新します。インメモリデータ構造への書き込みはcommit log書き込みの成功後にのみ実行されます。全てのcommit log書き込みはシーケンシャルなので、ディスクスループットを最大化するために我々は各マシンでcommit log専用のディスクを利用しています。インメモリデータ構造がデータサイズとオブジェクト数に基づく特定の閾値をまたいだ時、それ自体がディスクにダンプされます。この書き込みは各マシンが装備しているたくさんのコモディティなディスクの一つで行われます。全ての書き込みはディスクへのシーケンシャルと行キーベースの効率的なルックアップのためのインデックス生成です。これらのインデックスもまたデータファイルとともに永続化されます。やがてこのようなファイルがディスク上にたくさん存在するようになるので、バックグラウンドで単一のマージプロセスが走り、異なるファイルをつき合わせて一つのファイルにします。このプロセスはBigtableシステムで実行されるコンパクションプロセスととてもよく似ています。
典型的なread命令はディスク上のファイルを探す前にまずインメモリデータ構造を検索します。ファイルは新しいものから古いものの順で検索されます。ディスクルックアップが発生したとき、ディスク上の複数のファイルの中のキーを調べることがあります。キーを含まないファイルのルックアップを避けるために、ファイル中のキーを要約するブルームフィルタもデータとともに保存され、メモリ上にも保持されます。このブルームフィルタは探索されるキーがあるファイルの中に存在するかどうかをチェックするための最初の参考になります。カラムファミリー内のキーはたくさんのカラムを持つかもしれません。いくつかの特別なインデックスはキーからとても遠いカラムを取得する必要があります。ディスク上の全てのカラムをスキャンする事態を避けるため、我々はカラムを取得するのに正しいディスク上のチャンクにジャンプすることができるカラムインデックスを保持します。任意のキーのカラムはシリアライズされてディスクに書き出されるので、我々は256Kチャンク境界ごとにインデックスを生成します。この境界は設定可能ですが、我々の環境では256Kでよく動作することが分かっています。
単一マシン上でのCassandraプロセスは以下の概念から成っています：パーティショニングモジュール、クラスターメンバーシップおよび障害検出モジュールとストレージエンジンモジュールです。各モジュールは、メッセージパイプラインとタスクパイプラインがSEDA[20]アーキテク者のラインに沿って複数のステージに分割されたイベント駆動基板に依存しています。これらのモジュールは土台から最上部までJavaを使って実装されています。クラスタメンバーシップおよび障害検知モジュールはノンブロッキング I/Oを使うネットワークレイヤー上に構築されています。レプリケーションやリクエストルーティングのためのアプリケーション関連メッセージはTCPに依拠していますが、全てのシステム制御メッセージはUDPベースのメッセージングを用います。リクエストルーティングモジュールは若干の状態マシンを使って実装されています。read/writeリクエストがクラスター内のあるノードに到達した時、状態マシンは次の状態に変化します(i) キーのデータを所有するノード(たち)を特定する (ii)リクエストをそのノードにルーティングして到達のレスポンスを待つ (iii)設定されたタイムアウト値以内に返事が届かなかった場合、リクエストは失敗しクライアントに返す(iv)タイムスタンプに基づき最新のレスポンスを解明し(v)もしどれかのレプリカが最新のデータを持っていなかったら、データのリペアをスケジュールする簡単のため、我々は障害シナリオについてここでは語りません。システムは同期的にも非同期的にもwriteができるように設定出来ます。高スループットが要求されるような特定のシステムでは、我々は非同期レプリケーションを用います。ここではシステムに来るwriteはreadより遥かに多いです。同期的なケースでは結果をクライアントに返却する前にレスポンスのquorumを待ちます。
ジャーナルつきのシステムではcommit logエントリーをパージする機構の存在が必要です。Cassandraでは古いコミットログが特定の設定可能なサイズを超過したら新しいコミットログが起動するローリングコミットログを使っています。我々の環境では128MBサイズでのローリングコミットログがとても良いようです。各コミットログは基本的に固定長サイズのビットベクターのヘッダーを持ち、それは典型的には特定のシステムがハンドルするカラムファミリーの数より多いです。我々の実装では、カラムファミリごとに生成される単一のインメモリデータ構造と単一のデータファイルがあります。特定のカラムファミリーのためのインメモリデータ構造がディスクにダンプされると毎回、このカラムファミリーがディスクへの永続化を成功させたことを示すコミットログの中のビットがセットされます。これはこの情報の断片は既にコミット済みであることを表します。これらのビットベクターはコミットログ毎だけでなくメモリーにも保持されます。コミットログがロールされたとき、そのビットベクターと全ての先行してロールされていたコミットログのビットベクターがチェックされます。もし全てのデータがディスクに永続化されていることが確認されたら、コミットログは削除されます。コミットログへのwriteオペレーションはノーマルモードと高速同期モードがあります。高速同期モードでは、commit logへの書き込みはバッファされます。これは潜在的にマシンクラッシュでデータ消失するということです。このモードではまたインメモリデータ構造をディスクにバッファ方式でダンプします。伝統的なデータベースは特に高いwriteスループットをハンドルするようには設計されていません。Cassandraはディスクへの全ての書き込みをシーケンシャルライトに変えるので、ディスクへのwriteスループットを最大化します。ディスクへダンプされるファイルは変動しないので、読み込み時にロックが必要ありません。Cassandraのサーバーインスタンスはread/write命令に関して事実上ロックレスです。なので我々はB-Treeベースのデータベース実装に存在する並列性問題を扱ったりハンドリングしたりする必要はありません。
Cassandraシステムは全てのデータを主キーベースでインデックスします。ディスク上のデータファイルはブロックのシークエンスに分解されます。それぞれのブロックは多くても128キーを含み、ブロックインデックスによって境界を定められます。ブロックインデックスはブロック内の相対的なオフセットとそのデータのサイズをキャプチャします。インメモリデータ構造がディスクにダンプされた時、ブロックインデックスは生成され、それらのオフセットはインデックスとしてディスクに書き出されます。このインデックスは高速アクセスのためにメモリにも保持されます。典型的なread命令はいつも最初にインメモリデータ構造を検索します。もしあったらそのデータは、インメモリデータ構造にあるデータは全て最新なので、アプリケーションに返されます。もし見つからなかったら、時間の逆順にディスク上の全てのデータファイルに対してディスクI/Oを行います。我々は常に最新のデータを探しているので、最新のファイルを最初に探索し、データが見つかったらリターンします。時間とともにディスク上のデータファイルの数は増えていきます。我々はBigtableシステムととてもよくにたコンパクションプロセスを行い、複数ファイルを一つにマージします；主としてソート済みデータファイルの束をマージソートします。システムはいつもサイズの面で近いファイル同士を結合するので、100GBファイルと50GB以下のファイルを結合するような事態には陥りません。定期的にメジャーコンパクションプロセスが実行され、全ての関係データが一つの大きなファイルに結合されます。このコンパクションプロセスはディスクI/Oを酷使するオペレーションです。リードリクエストに影響を与えないようたくさんの最適化が導入されるるかもしれません。
Cassandraの設計、実装、保守の過程で、我々は多くの有用なな経験を得ていくつもの教訓を学びました。我々が学んだとても原則的な教訓のひとつは、アプリケーションによる利用の影響の理解なく新機能を追加しないこと、です。ほとんどの難しいシナリオは単なるノードクラッシュやネットワーク分断に根ざしたものではありません。いくつかの興味深いシナリオをここで紹介しましょう。
-
Inbox Searchアプリケーションをリリースする前、100M以上のユーザーのinboxデータ7TBをインデックスする必要があり、その時は我々のMySQL[1]インフラにストアされており、それをCassandraシステムに転送しました。全てのプロセスはMySQLデータファイルに対するMap/Reduce[7]ジョブを起動し、それらをインデックスしてCassandraに転置インデックスをストアしました。M/Rプロセスは実際にはCassandarのクライアントとして振る舞います。我々はいくつかのバックグラウンドチャンネルをユーザーごとの転置インデックスを結合しシリアライズしたデータを、シリアライゼーション/でシリアライゼーションのオーバーヘッドを避けるためにCassandraインスタンス間送信するM/Rプロセスに晒しましたが。この方法ではCassandraインスタンスはネットワーク帯域だけがボトルネックになりました。
-
殆どのアプリケーションはレプリカ単位のキーごとのアトミック操作しか要求しません。しかしながらいくつかのアプリケーションは主にセカンダリインデックスを保持する目的でトランザクショナルを要求してきました。何年もRDBMSでの業務を経験してきたほとんどのデベロッパーはそれがとても便利な機能だと考えました。我々はそのようなアトミック操作を提供するメカニズムに取り組んでいます。
-
我々は[15]や[5]で記述されたような多様な障害検知の実装を実験してきました。我々の経験では、障害検知するまでの時間はクラスターのサイズが成長するとともに許容出来る限界を超過します。100ノードのクラスターでのある実験では、ひとつノード障害を検知するまでにかかった時間は1分のオーダーでした。これは実際には我々の環境では動きません。やや保守的なPHIの値でのアクルーアル障害検知では、5に設定され、実験時では障害検出の平均時間は約15秒でした。
-
監視は当たり前のものだとは思われていません。Cassandraシステムは分散パフォーマンス監視ツールGanglia[12]とよく統合されています。我々は様々なシステムレベルのメトリクスをGangliaに提供し、これは我々のプロダクションワークロードが問題になったときにシステムの振る舞いを理解するの助けになります。ディスクははっきりした理由なく障害を起こします。ブートストラップアルゴリズムにはディスクが壊れたときにノードを修復するいくつかのフックがあります。これはしかし、運用上のオペレーションです。
-
Cassandraは完全に分散管理されたシステムですが、数個のコーディネーションを持つことはいくつかの分散機能をスムーズに実装するのに不可欠です。たとえばCassandraはZookeeperに統合されていますが、それは大規模な分散システムで様々なコーディネーションタスクを実行するのにも使えます。我々はZookeeperをCassandraをストレージエンジンとして利用するアプリケーションのじゃまを実際にしないいくつかの主要機能のために使うつもりです。
Inbox searchのため、メッセージの送信者と受信者の間で交わされた全てのメッセージのユーザーごとインデックスを保持します。現在有効になっている二種類の検索機能があり、(a)語句検索 (b)インタラクション - 人物の名前を貰いそのユーザーが送ったか受けとったかもしれないメッセージを全て返す - です。スキーマは二つのカラムファミリからなります。(a)クエリに関して、ユーザーIDはキーになり、メッセージを構成する語句はスーパーカラムになります。語句を含む各メッセージのメッセージ識別子はスーパーカラム内のカラムになります。クエリ(b)では、再びユーザーidがキーになり、受信者のIDがスーパーカラムになります。これらのスーパーカラムのそれぞれについてそれぞれのメッセージ識別子がカラムです。検索を速くするためにデータのインテリジェントキャッシングのための特定のフックを提供します。たとえば、ユーザーが検索バーをクリックしたとき、非同期メッセージがCassandraクラスターに送られ、ユーザーインデックスのバッファーキャッシュを準備します。この方法により、実際の検索クエリが実行されたとき検索結果はすでにメモリに乗っているかもしれません。システムは現在約50+TBのデータを150ノードのクラスターにストアしていて、それは西海岸と東海岸のデータセンター間に分かれています。我々はreadパフォーマンスのためにいくつかの調整されたプロダクションを持っています。
我々はスケーラビリティ、ハイパフォーマンス、そして幅広い適用性を提供するストレージシステムを作り、実装し、そして運用しました。我々はCassandraが低レイテンシにも関わらず非常に高い更新スループットを提供出来ることを経験的に証明しました。今後の活動は圧縮、キー間原子性サポートの能力、セカンダリインデックスのサポート追加を含みます。
CassandraシステムはFacebook内の多くの個人からのフィードバックにより多大な恩恵を受けました。加えて私たちの最初のプロダクションデプロイのためにMySQL内の全てのデータをインデックスしてCassandraに移行したKarthik Ranganathanに感謝します。また、EPFLから[19]や[8]に関する価値ある提言をくれたDan Dumitriuにも感謝します。
[1] MySQL AB. Mysql.
[2] Atul Adya, William J. Bolosky, Miguel Castro, Gerald Cermak, Ronnie Chaiken, John R. Douceur, Jon Howell, Jacob R. Lorch, Marvin Theimer, and Roger P. Wattenhofer. Farsite: Federated, available, and reliable storage for an incompletely trusted environment. In In Proceedings of the 5th Symposium on Operating Systems Design and Implementation (OSDI, pages 1–14, 2002.
[3] Mike Burrows. The chubby lock service for loosely-coupled distributed systems. In OSDI ’06: Proceedings of the 7th symposium on Operating systems design and implementation, pages 335–350, Berkeley, CA, USA, 2006. USENIX Association.
[4] Fay Chang, Jeffrey Dean, Sanjay Ghemawat, Wilson C. Hsieh, Deborah A. Wallach, Mike Burrows, Tushar Chandra, Andrew Fikes, and Robert E. Gruber. Bigtable: A distributed storage system for structured data. In In Proceedings of the 7th Conference on USENIX Symposium on Operating Systems Design and Implementation - Volume 7, pages 205–218, 2006.
[5] Abhinandan Das, Indranil Gupta, and Ashish Motivala. Swim: Scalable weakly-consistent infection-style process group membership protocol. In DSN ’02: Proceedings of the 2002 International Conference on Dependable Systems and Networks, pages 303–312, Washington, DC, USA, 2002. IEEE Computer Society.
[6] Giuseppe de Candia, Deniz Hastorun, Madan Jampani, Gunavardhan Kakulapati, Alex Pilchin, Swaminathan Sivasubramanian, Peter Vosshall, and Werner Vogels. Dynamo: amazonO ̃s highly available key-value store. In Proceedings of twenty-first ACM SIGOPS symposium on Operating systems principles, pages 205–220. ACM, 2007.
[7] Jeffrey Dean and Sanjay Ghemawat. Mapreduce: simplified data processing on large clusters. Commun. ACM, 51(1):107–113, 2008.
[8] Xavier D ́efago, P ́eter Urba ́n, Naohiro Hayashibara, and Takuya Katayama. The φ accrual failure detector. In RR IS-RR-2004-010, Japan Advanced Institute of Science and Technology, pages 66–78, 2004.
[9] Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung. The google file system. In SOSP ’03: Proceedings of the nineteenth ACM symposium on Operating systems principles, pages 29–43, New York, NY, USA, 2003. ACM.
[10] Jim Gray and Pat Helland. The dangers of replication and a solution. In In Proceedings of the 1996 ACM SIGMOD International Conference on Management of Data, pages 173–182, 1996.
[11] David Karger, Eric Lehman, Tom Leighton, Matthew Levine, Daniel Lewin, and Rina Panigrahy. Consistent hashing and random trees: Distributed caching protocols for relieving hot spots on the world wide web. In In ACM Symposium on Theory of Computing, pages 654–663, 1997.
[12] Matthew L. Massie, Brent N. Chun, and David E. Culler. The ganglia distributed monitoring system: Design, implementation, and experience. Parallel Computing, 30:2004, 2004.
[13] Benjamin Reed and Flavio Junquieira. Zookeeper.
[14] Peter Reiher, John Heidemann, David Ratner, Greg Skinner, and Gerald Popek. Resolving file conflicts in the ficus file system. In USTC’94: Proceedings of the USENIX Summer 1994 Technical Conference on USENIX Summer 1994 Technical Conference, pages 12–12, Berkeley, CA, USA, 1994. USENIX Association.
[15] Robbert Van Renesse, Yaron Minsky, and Mark Hayden. A gossip-style failure detection service. In Service,Tˇ Proc. Conf. Middleware, pages 55–70, 1996.
[16] Mahadev Satyanarayanan, James J. Kistler, Puneet Kumar, Maria E. Okasaki, Ellen H. Siegel, and David C. Steere. Coda: A highly available file system for a distributed workstation environment. IEEE Trans. Comput., 39(4):447–459, 1990.
[17] Ion Stoica, Robert Morris, David Liben-nowell, David R. Karger, M. Frans Kaashoek, Frank Dabek, and Hari Balakrishnan. Chord: a scalable peer-to-peer lookup protocol for internet applications. IEEE/ACM Transactions on Networking, 11:17–32, 2003.
[18] D. B. Terry, M. M. Theimer, Karin Petersen, A. J. Demers, M. J. Spreitzer, and C. H. Hauser. Managing update conflicts in bayou, a weakly connected replicated storage system. In SOSP ’95: Proceedings of the fifteenth ACM symposium on Operating systems principles, pages 172–182, New York, NY, USA, 1995. ACM.
[19] Robbert van Renesse, Dan Mihai Dumitriu, Valient Gough, and Chris Thomas. Efficient reconciliation and flow control for anti-entropy protocols. In Proceedings of the 2nd Large Scale Distributed Systems and Middleware Workshop (LADIS ’08), New York, NY, USA, 2008. ACM.
[20] Matt Welsh, David Culler, and Eric Brewer. Seda: an architecture for well-conditioned, scalable internet services. In SOSP ’01: Proceedings of the eighteenth ACM symposium on Operating systems principles, pages 230–243, New York, NY, USA, 2001. ACM.