Redis Streamsで時系列データをモデリングする方法

Redisは「データ構造サーバー」と呼ばれることが多く、シンプルなデータ構造プリミティブへのネットワークインターフェースを提供しています。Streams（ストリーム）は、何年も前にRedisがソート済みセット（Sorted Set）を導入して以来となる、初めての大型の汎用データ構造です。この記事では、この新しいデータ構造の主要な用途の一つである「時系列データのモデリング」について見ていきます。

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

Redis Streamsは、キーバリューペアの追記専用（append-only）の時系列を表現します。

任意の数のクライアントがStreamに書き込むことができ、書き込みのたびに、時系列に挿入されたアイテムに対する一意の昇順IDを受け取ります。

データを読み取るクライアントは、新しいデータの到着を待ちながらブロックすることもできますし、バッチ処理のために最後に読み取ったメッセージの「しおり（ブックマーク）」を保持することも、あるいはワークロードの共有やメッセージの確認応答（ACK）のために、より複雑な「コンシューマーグループ」を構成することも可能です。

Streamsは非常に大きなトピックなので、この記事では時系列データをRedisでモデル化する方法に絞って簡単に紹介します。従来のようにList型やSorted Set型で時系列を表現していた手法と比べると、この新しいデータ構造によってその実装は劇的にシンプルになりました。

ここでのコード例では、Node.js向けのioredisクライアントを使用します。

var Redis = require('ioredis');
var redis = new Redis();

Streamへのデータ送信

Streamsは追記専用であるため、通常必要な情報は書き込み先のRedisキーの名前と、キーバリューペアのセットだけです。

ここでは、空気質センサーからの測定値を記録し、site:pdxというキーにキーバリューペアのストリームを送信します。具体的には、現在の大気質指数（AQI）と摂氏温度を送信します。

redis.xadd('site:pdx', '*',
           'aqi', 37,
           'tempc', 5.1).then(function(id) {
  console.log("id:", id);
});

> 1527974818120-0

新しい測定値を記録するたびにXADDコマンドを送信します。レスポンスとして返されるのは、クエリに使用できる一意かつ常に昇順のIDです。IDの前半部分1527974818120はRedisサーバーが割り当てたタイムスタンプで、後半部分は複数のクライアントが同時に書き込んだ場合の衝突を避けるためのインクリメント番号です。

上記のように書き込みコマンドの第2引数に*を指定すると、Redis自身がタイムスタンプを決定してデータを保存します。

書き込み時にタイムスタンプを明示的に指定することも可能ですが、通常は推奨されません。Redisにタイムスタンプを選ばせることで、多数のクライアントがIDの選択や順序について調整することなく、単一のStreamへ同時に書き込めるようになります。そうした細部はRedisサーバーがクライアントに代わって処理してくれるのです。

シンプルな読み取り

いくつかの値がStreamに入ったら、IDまたはタイムスタンプの範囲を指定して値の範囲を取得できます。これは、空気質の最新の測定値をグラフに表示するアプリなどで活用できます。

redis.xrange('site:pdx',
             '1527974818120-0',
             '+',
             'COUNT', 5).then(function(resp) {

  // respには5件の読み取り値が格納される。グラフに渡す:
  // console.log(resp);
});

> [ [ '1543947167906-0', [ 'aqi', '31', 'tempc', '5.1' ] ],
> [ '1543947168312-0', [ 'aqi', '31', 'tempc', '5.3' ] ],
> [ '1543947168901-0', [ 'aqi', '31', 'tempc', '5.4' ] ],
> [ '1543947170033-0', [ 'aqi', '31', 'tempc', '5.4' ] ],
> [ '1543947171460-0', [ 'aqi', '31', 'tempc', '5.6' ] ] ]

Stream内の任意の場所から数値の範囲をサンプリングできるため、グラフ描画システムはパフォーマンスの劣化なしに過去のデータを照会できます。

ブロッキングとポーリング

データ範囲の照会はグラフ表示や履歴監視に便利ですが、データが到着した瞬間に即座に反応するシステムを作りたい場合もあります。そんなときにもRedis StreamsはXREADを使うことで理想的な選択肢となります。

redis.xread('BLOCK', 10000,
            'STREAMS', 'site:pdx', '$').then(function(resp) {

  // aqi > 50なら窓を閉める
  console.log(resp);
});

このようなブロッキング操作は、データが届くまで、あるいはタイムアウト（ここでは10,000ミリ秒）まで待機します。継続的なポーリングを維持するには、上記の例のようにデータが利用可能になるまでブロックし、データを受信するかコマンドがタイムアウトするたびに同じXREADコマンドを再度呼び出すだけで構いません。

Streamへの接続間でデータを取りこぼさないようにするには、読み取り時に最後に読み取ったIDを指定します。そうすれば、中断した箇所のちょうど続きから再開できます。

先ほどの例では、Streamに流れてくるすべてのデータが欲しかったため、「新着データのみ」を意味する特殊なトークン「$」を使用しました。IDを指定した場合のコマンドはほんの少し異なるだけです。

redis.xread('BLOCK', 10000,
            'STREAMS', 'site:pdx', '1543947171460-0');

また、複数の異なるStreamを一度に読み取ることもでき、データを受け付けた最初のStreamから値が返されます。

redis.xread('BLOCK', 10000,
            'STREAMS',
              'site:pdx', 'site:global',
              '1543947171460-0', '$');

この例では、site:pdxにID 1543947171460-0より新しいデータが書き込まれたとき、またはsite:globalに任意の新着データが書き込まれたときに、コマンドが結果を返します。

コンシューマー間の協調

大量のデータがStreamを流れるようになると、受信メッセージを処理するコンシューマーのコピーを複数持たせたくなるでしょう。これらのコンシューマーは、メッセージを取り出し、それに対してアクションを実行し、作業完了を「確認応答（ACK）」するという流れになります。Redis Streamsはこうした操作のためのプリミティブも備えています。詳細はここでは触れきれませんので、公式ドキュメントをご参照ください。

Streamコマンド早見表

Streams向けには多くの新しいRedisコマンドが用意されています。以下に簡単な一覧をまとめます。より詳しい情報は公式ドキュメントで確認できます。

基本コマンド：

• XADD：アイテム（キーバリューペアのまとまり）をStreamに追加する
• XRANGE / XREVRANGE：範囲を選択したり、Streamのアイテムを反復処理する
• XREAD：特定のIDより新しいアイテムを取得する（ブロック可）
• XTRIM：古いアイテムを破棄してStreamをトリミングする
• XDEL：Streamから特定のアイテムを削除する
• XLEN：Stream内のアイテム数を数える
• XINFO：Streamのメタデータを調査する

コンシューマーグループ関連コマンド：

• XGROUP：コンシューマーグループの作成・削除・リセット、およびメンバーの削除を行う
• XREADGROUP：XREADと同様だが、コンシューマーグループを使ってメッセージを受信する
• XPENDING：コンシューマーグループに配信済みだが未確認応答（未ACK）のメッセージを調査する
• XACK：コンシューマーグループに対してメッセージを確認応答し、ペンディングリストから削除する
• XCLAIM：停止したコンシューマーからメッセージを引き継ぐ

ユースケース

Streamsの最もよく引用されるユースケースはIoTワークロードです。センサーがStreamにデータを投入し、コンシューマーがそれをさまざまな用途（分析、コールドストレージへのアーカイブ、グラフ表示）に活用します。ここではサイズ上限付きのStreamが特に有効で、予測可能なメモリ使用量で固定サイズのStreamを確保できます。

また、Streamsは、以前他のRedisデータ構造を使っていたアプリケーションにも適しています。CeleryやSidekiqのようなキューアプリケーションは、Streamsのコンシューマーグループを活用することで、Redisネイティブな形で既読確認の検査を実現できます。さらに、Redis Pub/Subを使ったシンプルなチャットアプリを紹介するブログ記事は数多くありますが、Pub/Subはクライアントにパブリッシュされた後にメッセージを保持しないため、Redis Streamsを使えばより堅牢な実装にすることができます。

さっそく試してみませんか？

新しいRedisGreenサーバーなら、ほんの数クリックでRedis Streamsをお試しいただけます。