Streams: a new general purpose data structure in Redis.

2017-10-02

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

原文は Salvatore Sanfilippo により 2017-10-02 に公開されました。このブログを購読する

数ヶ月前まで、私にとってストリームは、メッセージングの文脈における興味深く比較的シンプルな概念以上のものではなかった。Kafkaがこの概念を広めてからは、主にDisqueというメッセージキューのケースでその有用性を検討してきた。Disqueは現在Redis 4.2のモジュールとして移植される予定のものだ。その後、DisqueはAP型のメッセージング、つまり耐障害性と、クライアントに大きな負担をかけずに配信を保証することに特化したものだと考えるようになり、ストリームという概念はそのケースにはあまり合わないと判断した。

しかし同時に、Redisがデフォルトで提供するデータ構造について、どうも落ち着かない問題があった。Redisのリスト、ソート済みセット、そしてPub/Subの機能の間には、ある種の隙間があるのだ。これらのツールをうまく使えば、メッセージやイベントのシーケンスをモデル化することはできるが、それぞれトレードオフがある。ソート済みセットはメモリを多く消費し、同じメッセージが何度も配信されるようなケースを自然にモデル化できず、クライアントは新しいメッセージを待ってブロックすることもできない。ソート済みセットはシーケンシャルなデータ構造ではなく、スコアを変えることで要素が移動するセットなのだから、時系列のような用途に適さないのも当然だ。リストにも別の問題があり、特定のユースケースで同様に適用が難しくなる。リストの中央にあるものを探索することはできない。なぜならその場合のアクセス時間は線形になるからだ。さらに、ファンアウトもできない。リストに対するブロッキング操作は、1つの要素を1つのクライアントにしか渡さない。リストには「この要素から先のものを取得してくれ」と指定するための固定の要素識別子もなかった。1対多のワークロードにはPub/Subがあり、多くのケースでは素晴らしいが、ファイア・アンド・フォーゲットでは困ることもある。履歴を残すことが重要なのだ。切断後にメッセージを再取得するためだけでなく、時系列のような一部のメッセージリストは、範囲クエリで探索することが非常に重要だからだ。「この10秒間の温度の測定値はどうだったか？」といった具合に。

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

誕生の経緯

Redis 4.0でモジュールが導入された後、ユーザー自身がこの問題を解決しようとし始めた。その一人であるTimothy Downs氏が、IRCで私にこう書き込んできた。

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

これが私の想像力をかき立てた。数日考えた末、これこそが上記の問題をすべて一度に解決できるチャンスかもしれないと気づいた。必要だったのは、「ログ」という概念を再び思い描くことだった。ログは基本的なプログラミング要素であり、ファイルを追記モードで開いて何らかの形式でデータを書き込むだけのシンプルなものなので、誰もが馴染みがある。しかし、Redisのデータ構造は抽象的でなければならない。データ構造はメモリ上にあり、RAMを使うのは楽をしたいからだけではなく、いくつかのポインタを使うことでデータ構造を概念化し抽象化して、明らかな制約から解放するためだ。たとえば通常のログにはいくつかの問題がある。オフセットは論理的なものではなく実際のバイトオフセットだが、エントリが挿入された時刻に関連する論理的なオフセットが欲しいとしたらどうだろう？そうすれば範囲クエリが無料で手に入る。同様に、ログはガベージコレクションが難しいことが多い。追記専用のデータ構造で古い要素をどうやって削除するのか？まあ、理想化されたログでは、単に「最大でこの数のエントリを保持したい」と言えば、古いものは消えていくのだ、といった具合だ。

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

いざ、コードを書く

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

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

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

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

> XADD mystream * sensor-id 1234 temperature 10.5
1506871964177.0

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

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

想像できる通り、2つのエントリを非常に高速に追加すると、シーケンス番号だけがインクリメントされることになる。この「高速挿入」は、MULTI/EXECブロックで簡単にシミュレートできる。

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

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

ストリームからデータを取得するには2つの方法がある。XRANGEコマンドで実装される範囲クエリと、XREADコマンドで実装されるストリーミングだ。XRANGEは単に開始から終了までの範囲のアイテムを取得する（両端を含む）。たとえば、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の範囲という観点で考えているが、IDの「シーケンス」部分を省略できるため、XRANGEを使って特定の時間範囲内の要素を取得することもできる。つまり、ミリ秒単位の時刻だけを指定できるのだ。次の例は「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 <name>」オプションを追加できるということだ。すると、同じグループ内のすべてのクライアントは*異なる*メッセージを受け取るようになる。もちろん、同じストリームから読み取る複数のグループが存在する可能性もあり、その場合、すべてのグループはストリームに到着する同じメッセージの複製を受け取るが、各グループ内ではメッセージは繰り返されない。

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

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

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

途中のアイテムの削除も可能にする予定だ。これは部分的にはすでに実装されているが、戦略としては、エントリフラグ内でエントリを削除済みとしてマークし、エントリと削除済みエントリの間の特定の比率に達したときにブロックを書き換えてガベージを回収し、必要に応じて断片化を避けるために隣接する別のブロックと結合するというものだ。

結論と提供時期

Redis streamsは年内にRedisの安定版4.0シリーズの一部となる予定だ。この汎用データ構造が、従来はカバーするのが難しく、現在のデータ構造を工夫して悪用しなければ特定の問題を解決できなかった多くのユースケースを、大きく補うことになると考えている。非常に重要なユースケースの一つが時系列だが、TREADを介した他のユースケース向けのメッセージストリーミングも、ファイア・アンド・フォーゲットよりも高い信頼性を必要とするPub/Subアプリケーションの代替としても、まったく新しいユースケースのためにも、非常に興味深いものになると感じている。とりあえず、自身の問題の文脈でこの新しい機能を評価し始めたいのであれば、GitHubで「streams」ブランチを取得して試してみてほしい。結局のところ、バグ報告は大歓迎だ :-)

動画がお好きなら、ストリームをリアルタイムで紹介するセッションはこちらだ：https://www.youtube.com/watch?v=ELDzy9lCFHQ