STREAMS Introduction

Redis 開発者教育の申込 Redis 定期点検/技術サポート
Redis Technical Support 		Redis エンタープライズサーバー
Redis Enterprise Server

紹介 Introduction

ストリーム Streams

ストリーム(Stream)はログデータを処理するために 5.0 で新たに導入されたデータ型です。
多くの産業(industry)ではデータが連続して発生します。 このデータの特徴は人ではなく機械(machine)が発生させ、連続かつ大量であることです。 もう一つの特徴は既存データを修正せず、追加だけで発生することです。 こうした種類のデータをストリーム(stream)またはログ(log)データと呼びます。
ストリームデータの実例として、 半導体、ディスプレイ、製鉄 などの製造工程では 温度、湿度、圧力、振動、傾き、 照度(明るさ)、煙 などを検知する多くのセンサー(sensor)が 動作し、秒以下の単位でデータを発生させます。 製造工程ではこのデータを使って製品の不良原因を分析し、 場合によっては不良を事前に予防することもできます。
半導体ウェハー(wafer)不良
身近な例をもう一つ挙げると、複数サーバーをモニタリングするとします。 各サーバーの CPU, Memory, Disk I/O, Network I/O などのデータを収集して リアルタイムチャートを表示し、アラームを送ることもできます。
Redis 5.0 ではこうしたデータの保管と処理に適したデータ構造であるストリーム(stream)を導入しました。
ディスク vs メモリ , RDB テーブル vs ストリーム
それではストリームにデータを入れ、照会する方法を一つずつ見ていきましょう。

データ投入(追加): XADD

Sensor-1234 の温度データを入れてみます。

XADD sensor-1234 * temperature 98.7
1538319053569-0

XADD の構文(syntax)は次のとおりです。

XADD key ID field value [field2 value2 ...]

key=sensor-1234, ID=*, field=temperature, value=98.7 です。
ここで注目すべきは ID=* です。ID を * で入力し、コマンドの結果として ID を返します。
ID は2つの数字で構成されます。
<millisecondsTime>-<sequenceNumber>
前は millisecond 単位の timestamp 数字で、 後ろは同じミリ秒内にさらにデータが入ると重複を防ぐため 1, 2, 3 のように1つずつ増加します。 この sequenceNumber は 8バイト整数なので、実質的に同一ミリ秒内に生成できるエントリ数に制限はありません。 ID を * で入力するのは Redis があるサーバー時刻で ID を生成することです。
ID を直接入力することもできます。

XADD sensor-1234 1538319053569-1 temperature 98.8
1538319053569-1

ID は以前に入力されたものより常に大きくなければなりません。同じか小さいとエラーが発生します。

データ長(項目数)の照会: XLEN

XLEN sensor-1234
(integer) 2

正確に言えば ID の個数を照会することです。

データ照会: XRANGE

XRANGE sensor-1234 - +
1) 1) 1538319053569-0
    2) 1) "temperature"
        2) "98.7"
2) 1) 1538319053569-1
    2) 1) "temperature"
        2) "98.8"

XRANGE の構文(syntax)は次のとおりです。

XRANGE key start end [COUNT count]

key=sensor-1234, start=-, end=+ です。
Count を使って照会するデータ件数を指定できます。
Start と end に特定 ID を指定できます。

XRANGE key 1538319053569 1538319053569

最近のデータから照会したい場合は XREVRANGE コマンドを使ってください。

XREVRANGE key + -

データを読む: XREAD

XREAD count 1 STREAMS sensor-1234 1538322045065-0
1) 1) "sensor-1234"
    2) 1) 1) 1538322045065-1
        2) 1) "temperature"
            2) "98.8"

XREAD の構文(syntax)は次のとおりです。

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
  • Count は読み取るデータ件数を指定します。Count を指定しなければすべてのデータを読みます。
  • ID は指定した ID の次のデータを読みます。最初のデータを読むには ID に 0 を指定します。
  • Block は新しいデータが入るのを待ってから読みます。このときは ID に特別に $ を使います。
    Lists の BLPOP と似ています。
XREAD block 5000 STREAMS sensor-1234 $

データ削除: XDEL, XTRIM

XDEL

XDEL sensor-1234 1538322045065-0
(integer) 1

XDEL の構文(syntax)は次のとおりです。

XDEL key ID

XTRIM

sensor-1234 にデータが 100 件あれば、古い順に 90 件を消し最新データ 10 件を残します。

XTRIM sensor-1234 maxlen 10

XTRIM の構文(syntax)は次のとおりです。

XTRIM key MAXLEN [~] count

特別なオプション ~ は約(about)です。   Sensor-1234 に100万件のデータがあれば 999,990 件を消すのに時間がかかります。 その間データ追加(XADD)や処理(XREAD)はできません。 大量データを迅速に処理すべきストリームでは、こうした処理遅延が起きないようにする必要があります。 短い時間で処理できる分だけデータを削除する機能です。

他のデータ型との比較

LIST

これまで Redis でこうしたストリーム/ログデータの処理に使われたデータ型は主に LIST でした。

  • LPUSH, XADD: 役割は似ています。速度差はあります。 LPUSH は O(1)、XADD は O(log(N)) です。   O(1), O(N), O(log(N))
  • RPOP, XREAD: RPOP はデータを読んで消します。XREAD は読むだけです。 RPOP は O(1)、XREAD は O(log(N)) です。 RPOP はデータを読んだ後にアプリケーションの問題で処理できなかった場合、復旧する方法がありません。
    LIST でデータを読むだけなら LINDEX コマンドを使ってください。
    STREAM で RPOP のように読んで消すなら XREAD と XDEL を使ってください。
  • LREM, XDEL: LREM は O(N)、XDEL は O(log(N)) です。
  • LTRIM, XTRIM: LTRIM は O(N)、XTRIM は O(log(N)) です。
  • 比較速度: LIST は値(string)を比較します。STREAM は ID(数字)を比較します。 そのため STREAM の方が比較が速いです。
  • データ: LIST はフィールドと値をアプリケーションで区別(パース)する必要があります。
    LSIT: "sensor-1234;1538319053569;temperature;98.7"
    STREAM は Redis 側で区別されます。
  • 途中へのデータ投入: LIST は可能です。STREAM は不可能です。

Sorted SET

Score/ID でソートされる点で2つのデータ型は似ています。 ストリーム/ログデータを ZSET に保存するなら score に timestamp を使います。

  • ZADD, XADD: どちらも O(log(N)) です。
  • ZPOPMAX, XREAD: 最後の1件を読むならどちらも O(log(N)) です。
  • ZREM, XDEL: どちらも O(log(N)) です。
  • データ: ZSET は値の重複を許しません。 そのため同じ値が来る場合に備えて、値に timestamp など区別できるデータを追加する必要があります。
    例) Score: 1538319053569, Value: "sensor-1234;1538319053569;temperature;98.7"
  • 途中へのデータ投入: ZSET は可能です。STREAM は不可能です。

HASH

フィールドを持つ点で2つのデータ型は似ています。ただし HASH で hgetall コマンドを実行すると入力されたフィールド順とは無関係に照会されますが、 STREAM では入力順を維持します。

他のデータ型にはないストリームの特徴・利点

  • Redis ストリームではコンシューマー(Consumer)を指定してデータを読め、
  • そのコンシューマーがデータを正しく処理したか確認する方法を提供し、
  • 正しく処理できていなければ他のコンシューマーに割り当てて処理させる方法を提供します。
    •

それではその処理方法を見ていきましょう。

コンシューマーグループを作る: XGROUP

XGROUP CREATE sensor-1234 ConsumerGroup $

テストデータを入れる

XADD sensor-1234 * temperature 100
XADD sensor-1234 * temperature 101
XADD sensor-1234 * temperature 102
XADD sensor-1234 * temperature 103
XADD sensor-1234 * temperature 104
XADD sensor-1234 * temperature 105

コンシューマーグループでデータを読む: XREADGROUP

XREADGROUP GROUP ConsumerGroup Consumer-A count 1 STREAMS sensor-1234 >
1) 1) "sensor-1234"
    2) 1) 1) 1538568726521-0
        2) 1) "temperature"
            2) "100"
XREADGROUP GROUP ConsumerGroup Consumer-A COUNT 1 STREAMS sensor-1234 >
1) 1) "sensor-1234"
    2) 1) 1) 1538568728545-0
        2) 1) "temperature"
            2) "101"

処理状況を確認する: XPENDING

XPENDING sensor-1234 ConsumerGroup
1) (integer) 2
2) 1538568726521-0
3) 1538568728545-0
4) 1) 1) "Consumer-A"
    2) "2"

処理を確定する: XACK

XACK sensor-1234 ConsumerGroup 1538568726521-0
(integer) 1

処理状況を再確認する

XPENDING sensor-1234 ConsumerGroup
1) (integer) 1
2) 1538568728545-0
3) 1538568728545-0
4) 1) 1) "Consumer-A"
    2) "1"

ペンディング時間を知る: XINFO CONSUMERS

XINFO CONSUMERS sensor-1234 ConsumerGroup
1) 1) name
    2) "Consumer-A"
    3) pending
    4) (integer) 1
    5) idle
    6) (integer) 31963

ペンディングデータを他のコンシューマーに割り当てて処理する: XCLAIM

XCLAIM sensor-1234 ConsumerGroup Consumer-B 30000 1538568728545-0
1) 1) 1538568728545-0
    2) 1) "temperature"
        2) "101"

このデータ1件についてだけ他のコンシューマーが処理するようにしたものです。
XCLAIM で処理したあとに XACK を送る必要があります。
XCLAIM の簡単な構文(syntax)は次のとおりです。

XCLAIM key group consumer min-idle-time ID

min-idle-time はミリ秒で指定します。
これからも続けてデータを読んで処理します。

XREADGROUP GROUP ConsumerGroup Consumer-B COUNT 1 STREAMS sensor-1234 >
1) 1) "sensor-1234"
    2) 1) 1) 1538568731409-0
        2) 1) "temperature"
            2) "102"

ストリーム1つに複数コンシューマーを割り当てる

1つのストリームで処理するデータが多く処理時間が長いなら、 次のようにコンシューマー(consumer)を複数指定できます。
例えば1秒に3件のデータ(Data-1,2,3)が入り、アプリケーションで1件の処理に1秒かかるなら 3つのコンシューマー(Consumer-A,B,C)を指定し、それぞれ別のアプリケーションで処理させます。

Data-1 -> Consumer-A
Data-2 -> Consumer-B
Data-3 -> Consumer-C
Data-4 -> Consumer-A
Data-5 -> Consumer-B
Data-6 -> Consumer-C

Streams コマンドリスト
CommandsVersionSyntaxDescription
XADD5.0.0key ID field string [field string ...]
XLEN5.0.0key
XRANGE5.0.0key start end [COUNT count]
XREVRANGE 5.0.0key end start [COUNT count]
XREAD5.0.0[COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
XDEL5.0.0key ID [ID ...]
XTRIM5.0.0key MAXLEN [~] count
XGROUP5.0.0[CREATE key group id-or-$] [DESTROY key group] [DELCONSUMER key group consumer] [SETID key group id-or-$]
XREADGROUP5.0.0GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
XACK5.0.0key group ID [ID ...]
XPENDING5.0.0key group [start end count] [consumer]
XCLAIM5.0.0key group consumer min-idle-time ID [ID ...] [IDLE ms] [TIME ms-unix-time] [RETRYCOUNT count] [FORCE] [JUSTID]
XAUTOCLAIM6.2.0key group consumer min-idle-time start [COUNT count] [JUSTID]
XINFO 5.0.0[CONSUMERS key group] [GROUPS key] [STREAM key] [HELP]

Total : 14


PDF manual

<< HRANDFIELD STREAMS Introduction XADD >>

Email 返事がかかってなれば、メールでお知らせします。
Email redis@redisgate.com Address Seoul Gangnam-gu Gangnam-daero 342 Yeoksam Bldg 5F Telephone 02.503.2235 Copyright © 2014-2026 redisGate
All right reserved