ElastiCache for Valkey 9.0の集計機能とは何か

2026年5月、AWSはAmazon ElastiCache for Valkey 9.0で集計機能（Aggregations）の一般提供を開始しました。これまでValkeyやRedisはキー・バリュー型のインメモリデータストアとして「高速な参照」を強みとしてきましたが、フィルタリングやグループ化、集計といった分析処理はアプリケーション側、あるいは外部のデータウェアハウスへ任せる構成が一般的でした。Aggregationsはこの分担を見直し、テラバイト規模のキャッシュ内データに対して単一のクエリで集計処理を完結させる仕組みを提供します。

AWSによると、Valkey 9.0はパイプラインスループットが従来比で最大40％向上しており、Aggregationsはこの基盤の上で動作します。ファセット検索のリアルタイムカウント更新、ゲームやストリーミングサービスのトレンド集計、商品カタログや在庫の運用レポートといったユースケースを、別途の分析基盤を構築せずインメモリで完結できる点が特徴です。料金面ではノードベースクラスターの利用料に含まれており、Aggregationsを使うために追加費用は発生しません。

対応リージョンは商用AWSリージョン全域に加え、AWS GovCloud（US）と中国リージョンを含みます。クライアントとしてはvalkey-py 6.1.1以降が必要で、Python 3.9以上で動作します。本記事では、機能の仕組みとPythonでの実装例、そして導入によって得られるアーキテクチャ上の利点を整理していきます。

パイプラインで理解するAggregationsの仕組み

Aggregationsは複数の処理ステージを連結したパイプラインとして実装されています。各ステージの出力が次のステージへの入力となり、最終的に集計結果が返されます。利用できる主なステージは次のとおりです。

パフォーマンスを引き出す鍵は、できるだけ早い段階でドキュメント数を減らすことです。FILTERや検索クエリを先頭で適用してデータ量を絞り、その後にGROUPBYやREDUCEを適用すれば、後段ステージの計算量を大幅に削減できます。プライマリノード上ではRead-after-Write整合性が保たれるため、直前の書き込みを反映した集計結果が得られる点も、リアルタイム性を要件とするアプリケーションにとって重要な特性です。

キャッシュ内で集計が完結する意義は単なる利便性にとどまりません。アプリケーションサーバーへ大量のドキュメントを転送して集計する従来構成と比べ、ネットワーク往復と転送量を大幅に削減できます。AWSは数百万件にまたがるカウント処理であってもマイクロ秒単位のレイテンシで結果が返ると説明しており、UI上でフィルタを切り替えるたびに数値を更新するような用途でも体感的な遅延が発生しにくい構造です。

主要ユースケースと設計の勘所

Aggregationsが活きる場面は、書き込み頻度が高く、かつ集計結果を即座に画面へ反映したい領域です。代表的な3つのユースケースを整理します。

ファセット検索は、ECサイトや検索画面で各フィルタ値のヒット件数をリアルタイムに表示する用途です。ユーザーがジャンルや価格帯を絞り込むたびに、残りのフィルタ候補ごとの件数が更新される体験を、別途の検索エンジン基盤を導入することなく実現できます。商品カタログの規模が大きくなっても、インメモリ集計のためページ応答時間への影響が抑えられます。

リアルタイムトレンドランキングは、動画ストリーミングやゲームプラットフォーム、マーケットプレイスでの活用が想定されます。直近24時間の視聴回数やプレイ回数といった指標をジャンル別に集計し、トップNを即時に返却するクエリを構成できます。バッチ処理で集計表を作る運用と比べ、書き込み直後のデータが反映される点が大きな違いです。

運用レポートでは、スタジオ別の作品数・平均評価・総視聴回数といった複合的な指標を、オンデマンドまたはスケジュール実行で算出できます。従来であれば別の分析クラスターへデータを連携する必要があった処理を、ElastiCacheのクラスター内で完結させられます。日次集計と即時集計を同じ仕組みでまかなえるため、運用面の構成要素を減らせる点も魅力です。

valkey-pyによる実装例

ここからはPythonクライアント（valkey-py）を用いた具体的なコード例を示します。事前にvalkey-py 6.1.1以降をインストールしておきます。クラスターエンドポイントへのSSL接続を前提とした初期化は次のとおりです。

import valkey
from valkey.commands.search.field import TextField, TagField, NumericField
from valkey.commands.search.indexDefinition import IndexDefinition, IndexType
from valkey.commands.search.aggregation import AggregateRequest, Desc
from valkey.commands.search import reducers

VALKEY_CLUSTER_ENDPOINT = "your-cluster.cnxa6h.clustercfg.use1.cache.amazonaws.com"

client = valkey.ValkeyCluster(
    host=VALKEY_CLUSTER_ENDPOINT,
    port=6379,
    decode_responses=True,
    ssl=True)

続いて作品カタログのインデックスを作成します。タグ型・数値型・テキスト型のフィールドを混在させ、ハッシュ型のキー（プレフィックス title:）に対してインデックスを定義します。

client.ft("catalog_index").create_index(fields=[
        TextField("title"),
        TagField("genre"),
        TagField("language"),
        TagField("studio"),
        NumericField("release_year"),
        NumericField("rating"),
        NumericField("views_24h")],
    definition=IndexDefinition(prefix=["title:"], index_type=IndexType.HASH))

インデックスが用意できれば、ファセット検索のカウントを取得するクエリは次のように記述できます。各次元ごとにグループ化し、COUNTリデューサで件数を集計する素直な構造です。

def get_facet_counts(filters):  # ファセット件数取得
    clauses = []
    if "genre" in filters:  # ジャンル指定があれば追加
        clauses.append(f"@genre:{{{filters['genre']}}}")
    if "language" in filters:  # 言語指定があれば追加
        clauses.append(f"@language:{{{filters['language']}}}")
    if "min_rating" in filters:  # 最低評価指定があれば追加
        clauses.append(f"@rating:[{filters['min_rating']} +inf]")
    query = " ".join(clauses) if clauses else "@rating:[-inf +inf]"

    facets = {}
    for dim in ["genre", "language", "rating"]:  # 次元ごとに集計
        req = AggregateRequest(query) \
            .load(f"@{dim}") \
            .group_by(f"@{dim}", reducers.count().alias("count"))
        facets[dim] = client.ft("catalog_index").aggregate(req).rows
    return facets

運用レポート用に複数のリデューサを組み合わせることも可能です。スタジオ別に作品数・平均評価・総視聴回数を一括で算出し、視聴回数の降順でソートする例を示します。

def get_studio_report():  # スタジオ別集計
    req = AggregateRequest("@rating:[-inf +inf]") \
        .load("@studio", "@rating", "@views_24h") \
        .group_by("@studio",
            reducers.count().alias("title_count"),
            reducers.avg("@rating").alias("avg_rating"),
            reducers.sum("@views_24h").alias("total_views")) \
        .sort_by(Desc("@total_views"))
    return client.ft("catalog_index").aggregate(req).rows

アーキテクチャ上の利点と導入時のポイント

Aggregationsを取り入れる最大の利点は、リアルタイム分析のためだけに専用の分析基盤を抱える必要がなくなる点です。書き込み直後のデータを反映した集計を、すでに運用しているElastiCacheクラスター内で実行できるため、構成要素の数とデータ連携の経路を減らせます。バックエンドエンジニアにとっては、ストリーミング処理基盤やETLパイプラインを設計せずに、まずキャッシュ内集計で要件を満たせるかを検討できる選択肢が増えたといえます。

導入時には以下の点に注意します。第一にValkey 9.0以降への対応が前提となるため、既存クラスターのバージョンアップ計画と整合させる必要があります。第二にAggregationsはインデックスに依存するため、検索対象とするフィールドの型（TagField、NumericField、TextField）を設計時に決めておくことが大切です。第三に大規模データセットではFILTERや検索クエリで早期に絞り込み、GROUPBYやREDUCEへ渡すドキュメント数を抑えることがレスポンスタイムの安定に直結します。

AWSの発表時点でAggregationsは商用全リージョン、AWS GovCloud（US）、中国リージョンで利用可能です。料金はノードベースクラスターの利用料に含まれます。既存のElastiCacheワークロードを大きく作り変えずに、低レイテンシのリアルタイム分析を段階的に組み込みたいチームにとって、検証する価値の高いアップデートです。まずは小さなインデックスを作成し、ファセット検索や運用レポートのクエリを置き換えてみるところから始めると、効果を測りやすいでしょう。