Streams: a new general purpose data structure in Redis.

2017-10-02

Streams：Redisの新しい汎用データ構造

数か月前まで、私にとってストリームは、メッセージングという文脈では興味深いとはいえ比較的わかりやすい概念に過ぎませんでした。Kafkaがこの概念を広めてからは、主にDisqueというメッセージキューにおける有用性を検証していました。Disqueは現在、Redis 4.2のモジュールとして移植が進められています。その後、DisqueはAP型のメッセージング、すなわちクライアント側に大きな負担をかけずに耐障害性と配信保証を実現することに主眼があると考えるようになり、ストリームという概念はあの場合には適さないと判断しました。

しかし同時に、Redisが標準で提供するデータ構造には、どうも落ち着かない問題がありました。Redisのリスト、ソート済みセット、そしてPub/Subの機能の間には、ある種の隙間があるのです。これらのツールを組み合わせれば、メッセージやイベントの連続をモデル化することはできますが、いずれも一長一短があります。ソート済みセットはメモリを多く消費し、同じメッセージを何度も配信するということを自然に表現できませんし、クライアントが新しいメッセージを待ってブロックすることもできません。ソート済みセットはそもそもシーケンシャルなデータ構造ではなく、スコアを変えることで要素を移動させられる集合ですから、時系列のような用途に適さないのも当然です。リストにも別の問題があり、特定のユースケースでの適用を難しくしています。リストの途中に何があるのかを覗こうとすると、アクセス時間が線形にかかってしまいます。さらにファンアウトもできず、リストに対するブロック操作では一つの要素が一つのクライアントにしか渡されません。リストには「この要素から先をください」と指定するための固定的な要素IDもありませんでした。一対多のワークロードにはPub/Subがあり、多くの場面で素晴らしく機能しますが、fire-and-forgetでは困ることもあります。履歴を残すことは重要です。切断後の再取得のためだけでなく、時系列のようなメッセージの列自体が範囲検索で調べることに意味があるからです。「この10秒間の温度の読み取り値はどうだったか？」といった具合に。

上記の問題に対処するために、私はソート済みセットとリストを一つのより柔軟なデータ構造に一般化しようと計画しました。しかし設計を試みるたびに、できあがるデータ構造は今あるものよりもはるかに不自然なものになってしまいました。Redisの良いところの一つは、公開されているデータ構造が「サルヴァトーレが発明したAPI」というものではなく、自然な計算機科学のデータ構造に近いことです。結局、私は試行をやめ、「今提供できるのはここまでだ。いずれPub/Subに履歴を足すか、リストのアクセスパターンにもう少し柔軟性を持たせることにしよう」と考えることにしました。しかし、カンファレンスでユーザーから「Redisで時系列をどうモデル化すればいいですか？」といった類の質問を受けるたびに、私の顔は青ざめていました。

誕生の経緯

Redis 4.0でモジュールが導入された後、ユーザー自身がこの問題を解決しようとし始めました。その一人であるTimothy Downs氏から、IRCで次のようなメッセージをもらいました。

<forkfork> 僕が作ろうとしているモジュールは、トランザクションログのようなデータ型を追加するものです。つまり、Redisのメモリ使用量を大きく増やさずに、非常に多くのサブスクライバーがPub/Subのようなことをできるようにするものです
<forkfork> Redis側で各コンシューマーがどこまで読んだかを管理したり、サブスクライバーごとにメッセージを複製したりするのではなく、サブスクライバー自身がメッセージキュー内の現在位置を保持する形です

このアイデアに私は心を掴まれました。数日考えた末、これこそが上記の問題を一度に解決する好機になり得ると気づいたのです。必要だったのは、「ログ」という概念を再び思い描くことでした。ログは基本的なプログラミング要素であり、誰もが親しんでいます。ファイルを追記モードで開き、何らかの形式でデータを書き込むだけのシンプルなものだからです。しかしRedisのデータ構造は抽象的でなければなりません。インメモリであり、RAMを使うのは決して手抜きだからではなく、わずかなポインタを使うことでデータ構造を概念化し、抽象化して、明白な制約から解放するためです。たとえば通常のログにはいくつかの問題があります。オフセットは論理的なものではなく実際のバイトオフセットです。もしエントリが挿入された時刻に関連した論理オフセットが欲しいとしたらどうでしょう。範囲検索が無料で手に入ります。同様に、ログはガベージコレクションが難しいことがよくあります。追記専用のデータ構造で古い要素をどう削除するのか。そう、私たちの理想化されたログでは、単に「最大でこの件数まで」と指定すれば、古いものは自動的に消えていく、といった具合です。

Timothy氏のアイデアを種に仕様を書き始めたちょうどその頃、私はRedis Clusterの内部の一部を最適化するために基数木（ラディックスツリー）の実装に取り組んでいました。これが、非常に省メモリでありながら、範囲取得を対数時間で可能にする効率的なログを実装するための土台となりました。同時に、設計にうまく取り入れられるアイデアはないかとKafka Streamsについても読み始め、その結果Kafkaのコンシューマーグループという概念を取り入れ、Redisとインメモリというユースケース向けに再び理想化しました。しかし仕様は数か月もの間ただの仕様のままで、最終的にはこのRedisへの追加機能について様々な人と話す中で得た多くのヒントを反映させるため、ほぼ一から書き直すほどでした。私はRedis Streamsを、他のイベントやメッセージング用途だけでなく、とりわけ時系列のユースケースにとって非常に優れたものにしたかったのです。

いよいよコーディングへ

Redis Confから戻った夏の間、私は「listpack」というライブラリを実装していました。このライブラリはziplist.cの後継にあたるもので、単一のアロケーションの中に文字列要素のリストを表現できるデータ構造です。非常に特化したシリアライズ形式であり、右から左へ、逆順にも解析できるという特徴があります。これはあらゆるユースケースでziplistを置き換えるために必要な特性でした。

基数木とlistpackを組み合わせることで、きわめて省メモリでありながらインデックス化された、つまりIDや時刻によるランダムアクセスが可能なログを簡単に構築できます。これが用意できたところで、ストリームデータ構造を実装するコードを書き始めました。実装はまだ仕上げの途中ですが、この時点でGitHub上のRedisの「streams」ブランチには、遊び始めるのに十分なものが揃っています。APIが100%確定したとは申しませんが、興味深い事実が二つあります。一つは、この時点で欠けているのはコンシューマーグループと、ストリームを操作するための重要度の低い幾つかのコマンドだけで、大きな機能はすでに実装されているということです。もう一つは、ストリーム関連の作業をすべて、約2か月後に4.0ブランチへバックポートすることを決めたことです。すべてが安定したと判断できれば、という条件付きですが。これは、Redisユーザーがストリームを使うためにRedis 4.2を待つ必要がなく、できるだけ早く本番利用できるようになることを意味します。新しいデータ構造であるため、コード変更のほとんどが新しいコードの中に自己完結しているので、これが可能になっています。唯一の例外はリストのブロック操作です。ストリームとリストのブロック操作で同じコードを共有するようにリファクタリングしたことで、Redis内部はむしろ大きくシンプルになりました。

チュートリアル：Redis Streamsへようこそ

ある意味、ストリームはRedisのリストを大幅に強化したものと考えることができます。ストリームの要素は単なる一つの文字列ではなく、フィールドと値からなるオブジェクトです。範囲検索が可能で、しかも高速です。ストリーム内の各エントリはIDを持ち、これは論理的なオフセットです。異なるクライアントは、指定したIDより大きいIDを持つ要素をブロックして待つことができます。Redis Streamsの基本的なコマンドはXADDです。そう、Redisのストリームコマンドはすべて「X」で始まります。

> XADD mystream * sensor-id 1234 temperature 10.5
1506871964177.0

XADDコマンドは、指定したエントリを新しい要素として指定のストリーム「mystream」に追記します。上記の例では、エントリはsensor-idとtemperatureという二つのフィールドを持っていますが、同じストリーム内の各エントリは異なるフィールドを持っていてもかまいません。同じフィールド名を使えば、メモリ使用量がより節約されます。興味深いことに、フィールドの順序も保持されることが保証されています。XADDは挿入されたばかりのエントリのIDを返します。3番目の引数にアスタリスクを指定して、IDの自動生成を要求したからです。これはほぼ常に望む動作ですが、特定のIDを強制することも可能です。たとえばスレーブやAOFファイルへコマンドをレプリケートする場合などです。

IDは二つの部分から構成されます。ミリ秒単位の時刻とシーケンス番号です。1506871964177がミリ秒時刻であり、単なるミリ秒精度のUnix時刻です。ドットの後の0がシーケンス番号で、同じミリ秒内にエントリが追加された場合に区別するために使われます。どちらも64ビットの符号なし整数です。つまり、同じミリ秒内であっても、好きなだけエントリをストリームに追加できます。IDのミリ秒部分は、Redisサーバーの現在ローカル時刻とストリーム内の最後のエントリのうち、大きい方が使われます。そのため、たとえコンピューターの時計が戻ったとしても、IDは増加し続けます。ある意味、ストリームエントリのIDは128ビットの数値全体として考えることができます。しかし、IDがエントリが追加されたインスタンスのローカル時刻と相関しているという事実は、ミリ秒精度の範囲検索が無料で手に入ることを意味します。

お察しのとおり、非常に高速に二つのエントリを追加すると、シーケンス番号だけが増加します。「高速挿入」は、MULTI/EXECブロックで簡単に再現できます。

> MULTI
OK
> XADD mystream * foo 10
QUEUED
> XADD mystream * bar 20
QUEUED
> EXEC
1) 1506872463535.0
2) 1506872463535.1

上の例は、最初にスキーマを定義しなくても、異なるエントリで異なるフィールドを使えることも示しています。実際には、各ブロック（通常50〜150件程度のメッセージを含みます）の最初のメッセージが参照用として使われ、同じフィールドを持つ後続のエントリは「このブロックの最初のエントリと同じフィールドです」という単一のフラグで圧縮されます。そのため、後続のメッセージで同じフィールドを使えば、フィールドの集合が時間をかけてゆっくり変化する場合でも、多くのメモリを節約できるのです。

ストリームからデータを取得するには二つの方法があります。XRANGEコマンドで実現される範囲検索と、XREADコマンドで実現されるストリーミングです。XRANGEは、開始（start）から終了（stop）まで、両端を含む範囲の要素を取得します。たとえば、IDが分かっていれば、単一の要素を次のように取得できます。

> XRANGE mystream 1506871964177.0 1506871964177.0
1) 1) 1506871964177.0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "10.5"

しかし、特別な開始記号である「-」と、特別な終了記号である「+」を使って、取りうる最小および最大のIDを表すこともできます。COUNTオプションを使って返すエントリ数を制限することも可能です。より複雑なXRANGEの例を次に示します。

> XRANGE mystream - + COUNT 2
1) 1) 1506871964177.0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "10.5"
2) 1) 1506872463535.0
   2) 1) "foo"
      2) "10"

ここまではIDの範囲という観点で考えてきましたが、XRANGEを使えば特定の時間範囲にある要素を取得することもできます。IDの「シーケンス」部分を省略できるからです。つまり、ミリ秒単位の時刻だけを指定すればよいのです。次の例は、「Unix時刻1506872463から始まる10件のエントリをください」という意味になります。

127.0.0.1:6379> XRANGE mystream 1506872463000 + COUNT 10
1) 1) 1506872463535.0
   2) 1) "foo"
      2) "10"
2) 1) 1506872463535.1
   2) 1) "bar"
      2) "20"

XRANGEについて最後に押さえておきたい重要な点は、返信の中にIDが含まれており、直後のIDはIDのシーケンス部分を単にインクリメントするだけで簡単に得られるため、XRANGEを使ってストリーム全体をインクリメンタルに反復することが可能だということです。指定した数の要素を呼び出しごとに受け取りながら、順に辿っていけます。Redisの*SCAN系コマンドが、反復処理向けに設計されていなかったにもかかわらずRedisのデータ構造の反復を可能にした後、同じ過ちを繰り返すことは避けました。

XREADによるストリーミング：新しいデータをブロックして待つ

XRANGEは、IDや時刻で範囲を取得したり、IDで単一の要素を取得したりしたい場合には最適です。しかし、異なるクライアントがデータの到着とともに消費していくようなストリームの場合、これだけでは十分ではなく、何らかのポーリングが必要になります（たまに接続してデータを取得するだけの特定のアプリケーションでは、それでも良いかもしれません）。

XREADコマンドは、最後に取得したストリーム内のエントリのIDを指定するだけで、複数のストリームから同時に読み取れるように設計されています。さらに、データがなければブロックして、データが到着したらブロックが解除されるように要求できます。リストのブロック操作で起こることと似ていますが、ここではデータはストリームから消費されず、複数のクライアントが同じデータに同時にアクセスできます。

これはXREAD呼び出しの典型的な例です。

> XREAD BLOCK 5000 STREAMS mystream otherstream $ $

これは次のような意味です。「mystream」と「otherstream」からデータを取得してください。データがなければ、5000ミリ秒のタイムアウトでクライアントをブロックしてください。STREAMSオプションの後には、リッスンしたいキーと、最後に持っているIDを指定します。ただし、特別なIDである「$」は、「今ストリームにあるすべての要素は持っているものとみなし、次に到着する要素からだけ取得してください」という意味になります。

もし別のクライアントから次のコマンドを送ると、

> XADD otherstream * message "Hi There"

XREAD側では次のようになります。

1) 1) "otherstream"
   2) 1) 1) 1506935385635.0
         2) 1) "message"
            2) "Hi There"

データを受信したキーと、受信したデータを一緒に取得できます。次の呼び出しでは、おそらく最後に受信したメッセージのIDを使うことになります。

> XREAD BLOCK 5000 STREAMS mystream otherstream $ 1506935385635.0

といった具合です。ただし、この使い方では、クライアントがメッセージの処理に時間がかかったなどの理由で、再接続までに非常に長い時間が空く可能性があります。その間、大量のメッセージが溜まってしまうかもしれません。そのため、XREADでは常にCOUNTオプションを使うのが賢明です。そうすれば、クライアントがメッセージであふれたり、サーバーが単一のクライアントのために大量のメッセージを提供するのに時間を取られたりすることを防げます。

長さ制限付きストリーム

ここまでは順調です……しかしストリームもいずれ古いメッセージを削除しなければなりません。幸い、これはXADDコマンドのMAXLENオプションで可能です。

> XADD mystream MAXLEN 1000000 * field1 value1 field2 value2

これは基本的に次のような意味です。新しい要素を追加した結果、ストリームのメッセージ数が100万件を超えた場合、古いメッセージを削除して長さを100万件に戻してください。リストでRPUSH + LTRIMを使うのと似ていますが、今回はそれを組み込みの仕組みとして備えています。ただし、上記の方法では、新しいメッセージを追加するたびに、ストリームの反対側からメッセージを削除する作業も発生し、CPUを消費することに注意してください。そのため、MAXLENの数値の前に「~」記号を使うことで、厳密に100万件ちょうどでなければならないわけではなく、少し多くても構わない、と指定することができます。

> XADD mystream MAXLEN ~ 1000000 * foo bar

このようにすると、XADDはノード全体を削除できる場合にのみメッセージを削除します。これにより、長さ制限付きストリームを持つコストは、通常のXADDと比べてほぼ無視できるものになります。

コンシューマーグループ（開発中）

これは、まだRedisに実装されていない機能の最初のものです。現在開発が進められています。Kafkaに最も明確に着想を得たアイデアですが、ここではかなり異なる形で実装されています。要点は、XREADで「GROUP <名前>」オプションも追加できるということです。同じグループに属するすべてのクライアントは、自動的にお互いに異なるメッセージを受け取ります。もちろん、同じストリームから読み取る複数のグループが存在する可能性もあり、その場合すべてのグループはストリームに到着する同じメッセージの複製を受け取りますが、各グループ内ではメッセージが重複することはありません。

グループの拡張として、グループを指定する際に「RETRY <ミリ秒>」オプションを指定できるようになる予定です。この場合、XACKで処理が確認応答されないメッセージは、指定されたミリ秒数の後に再び配信されます。これにより、クライアント側でメッセージを処理済みとしてマークする独自の手段がない場合でも、メッセージ配信にある程度の信頼性を提供します。この部分も現在開発中です。

メモリ使用量と保存・読み込み時間

Redis Streamsのモデルに用いられている設計のおかげで、メモリ使用量は驚くほど少なく抑えられています。使用量はフィールド数や値、その長さに依存しますが、シンプルなメッセージであれば100MBあたり数百万件のメッセージを格納できます。さらに、このフォーマットはシリアライズを最小限で済ませられるように考えられています。基数木のノードとして格納されるlistpackブロックは、ディスク上とメモリ上で同じ表現を持つため、保存や読み込みがきわめて簡単です。たとえばRedisは、RDBファイルから500万件のエントリを0.3秒で読み込むことができます。これにより、ストリームのレプリケーションや永続化は非常に効率的です。

将来的には、途中にある要素の削除も可能にする予定です。これは部分的にはすでに実装されていますが、戦略としては、エントリをエントリフラグの中で削除済みとしてマークし、エントリ数と削除済みエントリ数の比率が一定に達したときに、ブロックを書き直してガベージを回収し、必要に応じて断片化を避けるために隣接するブロックと結合するというものです。

結論と提供予定

Redis Streamsは、年内にRedis 4.0系統の安定版に含まれる予定です。この汎用データ構造によって、これまでカバーするのが難しかった多くのユースケースをRedisが広くカバーできるようになると考えています。つまり、これまでは既存のデータ構造を工夫して流用しなければ解決できなかった問題です。非常に重要なユースケースの一つは時系列ですが、私の感覚では、TREADを介した他のユースケース向けのメッセージストリーミングも、fire-and-forgetよりも高い信頼性を必要とするPub/Subアプリケーションの置き換えとしても、まったく新しいユースケースのためにも、非常に興味深いものになるでしょう。今のところ、ご自身の問題領域でこの新機能を評価し始めたいのであれば、GitHubで「streams」ブランチを取得して、ぜひ触ってみてください。バグ報告はもちろん歓迎します :-)

動画がお好きな方は、ストリームを実際に動かしているリアルタイムセッションがこちらにあります： https://www.youtube.com/watch?v=ELDzy9lCFHQ