RavenDB の Replication, Scale Out (Sharding)

前回 紹介した RavenDB について、続きを記載しようと思う。

今回は、RavenDB を使った Failover と Scaling について記載しておく。RavenDB における Scaling では、Scale Up ではなく、後述する Shared data set による Scale Out の手法が採用されている。

まずは、本題に入る前に、いくつか準備をしておこう。

準備 (RavenDB の HTTP ホスト)

今回は、複数の RavenDB サービスを起動して実験するので、RavenDB を HTTP ホストで起動する。(実行ファイルを展開して、起動する。)
まず、RavenDB の Build を ダウンロード して、ダウンロードした zip を展開する。 つぎに、RavenDB のインストールと起動をおこなう。RavenDB をインストール (実行) するには、インストール フォルダに移動して、以下のコマンドを実行する。(/uninstall で簡単にアンインストールできる。)

.\Server\Raven.Server.exe /install

上記のコマンドを実行すると、下図の通り、Windows のサービスが登録されて起動する。(次回から、OS の起動の際に、自動で起動する。)

なお、今回、データベースに対してどのような要求が渡されたか確認するため、 デバッグ モードで実行してみる。 デバッグ モードで実行するには、上記のコマンドではなく、 下記のコマンドを実行する。
デバッグ モードの場合、上記のような Windows サービスではなく、実行したコンソール上で HTTP のプロセスが実行され、どのような要求を処理したかコンソール上に表示されるようになる。

cd .\Server
Raven.Server.exe –debug

なお、既定では、ポート 8080 で起動する。(起動しているかどうかは、ブラウザーで、http://localhost:8080/raven/studio.html にアクセスしてみると良い。) このポート番号を変更するには、 Server\Raven.Server.exe.config を開いて、下記の通り Raven/Port を変更すれば良い。(複数の RavenDB を同じマシンで起動するには、このポートを変更して起動すれば良い。)

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <appSettings>
    <add key="Raven/Port" value="8081"/>
    <add key="Raven/DataDir" value="~\Data"/>
    <add key="Raven/AnonymousAccess" value="Get"/>
  </appSettings>
  . . .
</configuration>

RavenDB を管理するには、下記の URL に接続して、RavenDB Management Studio というブラウザー インターフェイスを利用すると便利だ。(プログラムでも管理できるけど、このほうが超便利。)

http://<install server>:<port>/raven/studio.html

例によって、Data フォルダー (.\Server\Data) を消すとデータは初期化され、ポータブルな運用ができるので、いろいろ作ってみては、データを消して試すことができる。

以降のサンプル コードでは RavenDB Client (.NET の API) を使ってアクセスをおこなうが、HTTP ホストの場合、クライアント側は、API を使わず、HTTP をそのまま呼び出して、RESTful な方法でデータ アクセスができるため、jquery などを使って実装しても良い。

RavenDB への Plug-in (Bundles)

RavenDB では、プラグイン (Plug-in) 可能な追加の機能を bundle と呼んでおり、 後述する Replication でも、この bundle を使用する。 bundle の追加は非常に簡単で、RavenDB を実行するディレクトリの下に Plugins フォルダーを作成し、ここに必要な dll を配置するだけだ。(このため、 RavenDB を起動する場所によって Plugins フォルダーの場所が変わるので注意。一般には、RavenDB の実行モジュールが入っている .\Server の下に作成しておけば良い。)

インターネット上から bundle をダウンロードして Plugins フォルダーに配置するためのスクリプトが用意されていて、 例えば、今回使用する Replication bundle をプラグインするには、PowerShell を管理者権限で起動し、下記の通り実行する。

# enable script execution
Set-ExecutionPolicy Unrestricted

# download Raven.Bundles.Replication.dll
#   to Plunins folder
cd .\Server
..\Raven-GetBundles.ps1 Replication

なお、上記の RavenDB の zip を展開すると、インストール フォルダーの Bundles フォルダーに、既にいくつかの bundle の dll が入っているので、 Plugins サブ フォルダーを作成し、ここに手動でコピーしても良い。(ここには、Replication bundle も入っている。)

Replication

では、早速、Replication から説明しよう。
以降では、サーバーを 8080、8081 の各ポートで 2 台起動していると仮定する。

まずは、使用するすべてのサーバーで、上述した Replication bundle がプラグイン (Plug-in) されていることを確認する。

つぎに、Replicatoin の構成をおこなう。Raven DB では、構成情報もドキュメント (Json ドキュメント) として登録するようになっていて、こうした管理用のドキュメントは System Document (Sys Doc) と呼ばれている。今回は、Replication 用のドキュメントを登録する。

ブラウザーを起動し、8080 のサーバーの RavenDB Management Studio (URL は上述) を使用して、 Documents タブを選択し、[Create a Document] をクリックして、 ドキュメントを作成する。
Key を「Raven/Replication/Destinations」として、下記の Json ドキュメントを追加する。

{
  "Destinations": [
    {
      "Url": "http://localhost:8081/"
    }
  ]
}

今回は、8080 のサーバーが master となり、8080 の更新を 8081 に Replication する。(そのため、8081 のサーバーに、上記の構成は必要ない。) 合計 2 台構成なので 1 台分の slave しか追加していないが、 3 台構成以上の場合は、上記の Json 配列に複数のマシンを追加すれば良い。

構成を変更したら、RavenDB を再起動する。 今回は、デバッグ実行しているので、「q」で抜けてから再度起動すれば良い。もしサービスとしてインストール (Windows サービスとして起動) している場合は、 管理者権限で以下のコマンドを実行すれば再起動できる。

Raven.Server.exe /restart

以上で、Replication の設定は完了だ。

では、実際にデータを更新して確認してみる。(8080 と 8081 のサーバーをデバッグ モードで起動しておこう。)

今回は、前回 と違って HTTP ホストの RavenDB を使うので、 クライアント側は RavenDB Client のみで充分だ。(NuGet からインストールできる。なお、前述の通り、jquery などを使ってアクセスしても良い。)
RavenDB Client を使って、下記の通りデータを登録してみる。(なお、前回のように、接続先の情報を .config に記述しておいても良い。)

using Raven.Client;
using Raven.Client.Document;

static void Main(string[] args)
{
  using (var ds = new DocumentStore
  {
    Url = "http://localhost:8080/"
  })
  {
    ds.Initialize();

    using (IDocumentSession session
        = ds.OpenSession())
    {
      Order o1 = new Order
      {
        Name = "test1",
        Price = 100,
        Category = "material"
      };
      session.Store(o1);
      session.SaveChanges();
      //session.Dispose();
    }

    Console.WriteLine("Done !");
    Console.ReadLine();
    //ds.Dispose();
  }
}

public class Order
{
  public string Name { get; set; }
  public int Price { get; set; }
  public string Category { get; set; }
}

8080 と 8081 のコンソールを見ながら、サーバー上でどんな要求が処理されているか確認すれば一目瞭然だ。8080 のサーバーに POST/PUT の要求がおこなわれると同時に、8081 のサーバーにも POST 要求が入る。その結果、Management Studio で見ると、双方に、同じデータが登録されているのが確認できる。

参考までに、8081 のサーバー上 (slave 上) のコンソール結果を出力すると下記の通りになる。

c:\Demo\RavenDB-8081\Server> Raven.Server.exe --debug
Raven is ready to process requests. Build 960, Version 1.0.0 / bce65ae
Server started in 3,298 ms
Data directory: c:\Demo\RavenDB-8081\Server\Data
HostName: <any> Port: 8081, Storage: Esent
Server Url: http://machine01:8081/
Available commands: cls, reset, gc, q

Request # 1: GET  - 614 ms - <default> - 200 - /replication/lastEtag?from=http%3A%2F%2Fmachine01%3A8080%2F&currentEtag=00000000-0000-0300-0000-000000000002
Request # 2: POST - 278 ms - <default> - 200 - /replication/replicateDocs?from=http%3A%2F%2Fmachine01%3A8080%2F
Request # 3: GET  - 266 ms - <default> - 200 - /replication/lastEtag?from=http%3A%2F%2Fmachine01%3A8080%2F&currentEtag=00000000-0000-0300-0000-000000000003
Request # 4: GET  -   5 ms - <default> - 200 - /replication/lastEtag?from=http%3A%2F%2Fmachine01%3A8080%2F&currentEtag=00000000-0000-0300-0000-000000000003
Request # 5: GET  -   3 ms - <default> - 200 - /replication/lastEtag?from=http%3A%2F%2Fmachine01%3A8080%2F&currentEtag=00000000-0000-0300-0000-000000000003

今回はテストのため 2 台としているが、単一のサーバー上で更新がおこなわれると、ファームのすべてのサーバーに更新のバッチが送信される。(この処理は、background で並列に処理される。)

また、今回は、8080 のサーバーを master として、このサーバーに発生した更新処理と同期する 8081 のサーバー (slave) を構成したが、 master – master の構成も可能だ。 この際、もし、サーバー間の更新が Conflict した場合は、下記のドキュメントの通り処理すれば良い。

[RavenDB] Dealing with replication conflicts
http://ravendb.net/docs/server/bundles/replicationconflicts

また、Failover の仕組みも提供している。
例えば、クライアントを下記の通り作成し、最初の ReadLine() の箇所で 8080 のサーバーを shutdown してみる。すると、以降の処理で例外は発生せず、データは、ちゃんと 8081 から取得される。(ただし、下記の Initialize() によって Replication Server の情報を読み込むので、Initialize の際に 8080 のサーバーが起動していなければならない。)

using (var ds = new DocumentStore
{
  Url = "http://localhost:8080/"
})
{
  ds.Initialize();

  Console.ReadLine(); // Wait and shutdown 8080 !!

  using (IDocumentSession session
      = ds.OpenSession())
  {
    Order item = session.Load<Order>("orders/1");
    Console.WriteLine("Price is ${0}.", item.Price);
  }

}

また、以下の通り記述すると、データ取得の際、8080 と 8081 のサーバーに交互に GET 要求が送信される。(この手法は、read striping と呼ばれている。)

using (var ds = new DocumentStore
{
  Url = "http://localhost:8080/",
  Conventions =
  {
    FailoverBehavior = FailoverBehavior.ReadFromAllServers
  }
})
{
  . . .

なお、master-master の場合は、上記で FailoverBehavior.AllowReadsFromSecondariesAndWritesToSecondaries を指定すると良い。

また、途中まで Replication を使わず実行し、 途中から Replication を構成した場合など、 サーバー間でデータの相違 (矛盾) が生じるように思われるかもしれないが、 ちゃんと、最初の同期処理でデータを同一に揃えてくれる。(初回に、slave に対し、同期するデータの回数分、POST 要求が送信される。)

ちなみに、RavenDB のドキュメント を見ると、 PutIndex、DeleteIndex は Replication でサポートされていないようなので注意してほしい。 なお、当然だが、Query をおこなうと、ちゃんと dynamic index は作成される。(ただし、その Index は Replication されない。対象のサーバーへ Query をおこなう度に、そのサーバーごとに Index が作成される。)

Sharding (Scale out using shared data set)

さて、いよいよ、RavenDB の Scale out の話に入りたい。

MongoDB 同様、RavenDB にも Sharding が提供されている。(Sharding とは、方針に沿って、データを複数のサーバーに分散すること。) 最新の RavenDB では、Sharding 環境で Indexing や Linq Query もサポートされている。
以降で、ちょっと詳しく見てみよう。

まず、単純に、データを任意に (おまかせで) 分散させる Blind Sharding を見てみよう。 これも、超簡単 ! 下記の通り、DocumentStore の代わりに、ShardedDocumentStore というオブジェクトを使えば完了だ。(bundle も不要。)
下記で、o1、o2 は、それぞれ別々のサーバーに振り分けられる。
つまり、Sharding は、サーバー側で実行されているのではなく、すべてクライアント側でおこなわれる。

using Raven.Client.Shard;

var stores = new Dictionary<string, IDocumentStore>
{
  {
    "server1",
    new DocumentStore {Url = "http://localhost:8080"}
  },
  {
    "server2",
    new DocumentStore {Url = "http://localhost:8081"}
  }
};

var shrd = new ShardStrategy(stores);

using (var ds =
  new ShardedDocumentStore(shrd))
{
  ds.Initialize();

  using (IDocumentSession session
      = ds.OpenSession())
  {
    Order o1 = new Order
    {
      Name = "test1",
      Price = 100,
      Category = "material"
    };
    session.Store(o1);
    session.SaveChanges();

    Order o2 = new Order
    {
      Name = "test2",
      Price = 200,
      Category = "material"
    };
    session.Store(o2);
    session.SaveChanges();
  }
}

つぎに、方針 (ポリシー) に沿って Sharding をおこなう Smart Sharding をプログラミングする。
まず、簡単な例として、Order の Category ごとに、データを別々のサーバーにわけて配置するサンプル コードを下記に記載する。
下記の場合、o1、o3 は 8080 のサーバーに配置され、o2 のみ 8081 のサーバーに配置される。

var stores = new Dictionary<string, IDocumentStore>
{
  {
    "material",
    new DocumentStore {Url = "http://localhost:8080"}
  },
  {
    "food",
    new DocumentStore {Url = "http://localhost:8081"}
  }
};
var shrd = new ShardStrategy(stores)
  .ShardingOn<Order>(o => o.Category);
using (var ds =
  new ShardedDocumentStore(shrd))
{
  ds.Initialize();

  using (IDocumentSession session
      = ds.OpenSession())
  {
    Order o1 = new Order
    {
      Name = "ball pointpen",
      Price = 100,
      Category = "material"
    };
    session.Store(o1);
    session.SaveChanges();

    Order o2 = new Order
    {
      Name = "ice cream",
      Price = 150,
      Category = "food"
    };
    session.Store(o2);
    session.SaveChanges();

    Order o3 = new Order
    {
      Name = "notebook",
      Price = 80,
      Category = "material"
    };
    session.Store(o3);
    session.SaveChanges();
  }
}

さらに、おもしろい実験をしてみよう。
例えば、Order と Product を下記の通り定義し、Order.Product に Product の Id を設定する。(前回 説明したように、RavenDB では、ドキュメントに、常に、Id が付与される。)

public class Order
{
  public string Id { get; set; }
  public string Product { get; set; }
  public int Count { get; set; }
}

public class Product
{
  public string Id { get; set; }
  public string Name { get; set; }
  public int Price { get; set; }
  public string Category { get; set; }
}

そして、下記の通り実行してみる。 すると、p1、p3、o1 は 8080 のサーバーに保存され、p2、o2 は 8081 のサーバーに保存される。

var stores = new Dictionary<string, IDocumentStore>
{
  {
    "material",
    new DocumentStore {Url = "http://localhost:8080"}
  },
  {
    "food",
    new DocumentStore {Url = "http://localhost:8081"}
  }
};
var shrd = new ShardStrategy(stores)
  .ShardingOn<Product>(p => p.Category)
  .ShardingOn<Order>(o => o.Product);
using (var ds =
  new ShardedDocumentStore(shrd))
{
  ds.Initialize();

  using (IDocumentSession session
      = ds.OpenSession())
  {
    // Create Product
    Product p1 = new Product
    {
      Name = "ball pointpen",
      Price = 100,
      Category = "material"
    };
    session.Store(p1);
    Product p2 = new Product
    {
      Name = "ice cream",
      Price = 150,
      Category = "food"
    };
    session.Store(p2);
    Product p3 = new Product
    {
      Name = "notebook",
      Price = 80,
      Category = "material"
    };
    session.Store(p3);
    session.SaveChanges();

    // Create Order
    Order o1 = new Order
    {
      Product = p3.Id,
      Count = 3
    };
    session.Store(o1);
    Order o2 = new Order
    {
      Product = p2.Id,
      Count = 2
    };
    session.Store(o2);
    session.SaveChanges();
  }
}

さて、勘の良いプログラマーなら そろそろ気づいたと思うが、Smart Sharding では、読み取り (検索) と連携することで効果を発揮する。実際に、その効果を見てみよう。

例えば、Load をおこなってみる。
まず、予備知識として、Sharding を使用した場合、Id の既定値は、前回 説明した <class name>/<sequence number> ではなく、<server name>/<class name>/<sequence number> となるので注意してほしい。
このため、例えば、p3 を取得する場合は、下記のプログラム コードになる。

Product item = session.Load<Product>("material/products/3");

さて、この際、サーバーにどのような処理が渡されたか、サーバー上のコンソール ウィンドウ (上述した debug 実行のコンソール) で確認してみてほしい。実は、この GET 要求は 8080 のサーバーにしか飛ばない。Id から、このオブジェクトが 8080 のサーバーにあることがわかっているためだ。

では、下記はどうだろう ? この Load でも、Order オブジェクトと、関連する Product オブジェクトは 8080 のサーバーにしかないため、8081 への問い合わせ (GET) はおこなわれない。下記の Include メソッドによって、8080 のサーバーへの 1 回の HTTP GET のみで結果を取得する。

// Prefetch Product object using Include
Order order = session.Include<Order>(o => o.Product)
  .Load("material/orders/1");
// This dosen't ask to server !
Product product = session.Load<Product>(order.Product);
Console.WriteLine("Price is ${0}. Count is {1}.",
  product.Price, order.Count);

Query でも同様だ。下記のサンプル コードの場合、検索や Index 作成は 8080 のサーバーでしかおこなわれず、余計なラウンドトリップは発生しない。

var q = from c in session.Query<Product>()
    where c.Category == "material"
    select c;
foreach (var item in q)
{
  Console.WriteLine("{0} : {1}",
    item.Name,
    item.Price);
}

and / or など複雑な query をおこなった場合も同様だ。今回は 2 台のサーバーだけで確認しているが、Linq Query の構文を解析し、m 台あるうちの n 台から結果を取得すれば良いと判断されると、 RavenDB Clinet は、その n 台のみに検索をおこない、取得したデータを結合して返してくる。(or を使った場合、など。)

まさに、Smart ! (クライアント ライブラリーが、賢いってことだね)

一方、下記の Query は、8080、8081 の双方のサーバーで実行されて、答えを返してくる。

var q = from c in session.Query<Product>()
    where c.Category == "notebook"
    select c;
foreach (var item in q)
{
  Console.WriteLine("{0} : {1}",
    item.Name,
    item.Price);
}

また、プログラマーなら、orderby を使った場合の動きが気になるよね。サーバー A とサーバー B で Index を使って Sort 結果を取得し、 最後に、これらを結合すると仮定すると、その結果、順番がばらばらになるような気がする。しかし、実際に、このような状態を作って orderby を実行してみると、 orderby の結果もちゃんと正しい答えが返ってくる。
きっと、ここは、RavenDB Client が頑張っちゃっているのかもしれない。(もしそうだとすると、この点は、データ数が多い場合に要注意ということだ。)

あと、Sharding Strategy を運用途中で変更すると、 その変更内容によっては、当然、検索結果はおかしくなってしまうので注意してほしい。(例えば、上記で、”material” と “food” のサーバーを入れ替えると、Query の際に誤った結果が返ってくることになる。) もちろん、それまで登録されていたデータと矛盾しない変更であれば問題ない。開始時点で、ちゃんと以降の運用も検討に入れて使ったほうが良さそうだ。

あと、Sharding と Replication の併用も可能だ。

いろいろ解説しはじめるときりがないが、これだけの事が、かなり安価に実現できる点は魅力的だ。(「費用」という意味ではなく、構成の理解やセットアップなど、トータルの意味で「安価」と書いている。)
他の RDB などでも、いまどき Replication や Distribution の仕組みくらいは持っているが、この手のデータベースの良い点は、速度はもちろんだが、とにかくポータブルで、「わかりやすい」という点だろう。動きがわかりやすいと、そのシステム固有の 訳のわからない動きに悩まされることも少ないし、判断も早い。

広告

RavenDB の Replication, Scale Out (Sharding)」への1件のフィードバック

  1. ピンバック: RavenDB の特徴と使い方 (プログラミング) « Net&Web

コメントは受け付けていません。