この記事は「2026 Japan AWS Jr. Champions 真夏のQiitaリレー」の5日目の記事となります。
過去の投稿（リンク集）と昨日投稿の記事は以下からご覧ください。
はじめに
位置情報から最寄り地点を探し、その地点に紐づく情報をさらに検索する処理を実装し、パフォーマンスを比較しました。
| 処理時間 | |
|---|---|
| 全件検索 | 743.8ms |
| ElastiCache Serverless for Valkey（ウォーム時） | 8.27ms |
ElastiCache Serverless for Valkeyを活用して約90倍検索を高速化できたので、その過程を共有します。
忙しい方向け
-
GEOADD/GEOSEARCHとZADD/ZRANGEBYSCOREによるインデックス化により高速化 - ClusterCrossSlotErrorはredis.cluster.RedisClusterと2,000件ごとのデータ投入により解消
位置情報の全件検索を再現
以下の流れで全件検索する実装をおこないました。
- 現在地から一番近い地点を探す
- ①で見つかった地点に紐づく情報（時刻表）をさらに検索する
この全件検索の流れは、N+1問題と構造的に同じです。
1件の検索結果を得るたびに、追加でもう一段クエリが発生しています。
さらに今回は、その追加クエリ自体がテーブル全件をなめる処理だったため、影響がより顕著に出ました。
実装
題材として、公共交通オープンデータ（GTFS-JP形式）から、バス停座標3,886件、時刻表データ592,164件を使いました。
python
def find_nearest_stop_naive(current_lat, current_lon, path="stops.txt"):
nearest, nearest_dist = None, float("inf")
with open(path, encoding="utf-8-sig") as f:
for row in csv.DictReader(f):
d = haversine(current_lat, current_lon, float(row["stop_lat"]), float(row["stop_lon"]))
if d < nearest_dist:
nearest_dist, nearest = d, row
return nearest, nearest_dist
def find_next_departures_naive(stop_id, after_time, path="stop_times.txt"):
results = []
with open(path, encoding="utf-8-sig") as f:
for row in csv.DictReader(f):
if row["stop_id"] == stop_id and row["departure_time"] >= after_time:
results.append(row)
return sorted(results, key=lambda r: r["departure_time"])
実測結果
全件検索の実測結果は以下の通りです。
| 処理 | 時間 |
|---|---|
| 最寄り地点検索（3,886件フルスキャン） | 9.9ms |
| 紐づく情報の検索（592,164件フルスキャン） | 733.9ms |
| 合計 | 743.8ms |
件数の多いテーブルへのフルスキャンが、そのまま支配的なコストになっていることがわかります。
ElastiCache Serverless for Valkeyでインデックス化
全件検索による課題を解決するため、Amazon ElastiCache Serverless for Valkeyを使い、あらかじめインデックスを作っておく形に変更しました。
以下の構成で環境を構築しました。
Amazon ElastiCache Serverless for Valkeyとは？
Valkeyは2024年にRedisのライセンス形態が変更されたことをきっかけに生まれた、Redis OSS互換のインメモリKVSです。
コマンド体系やデータ構造はRedisと共通しているため、redis-pyのような既存のRedisクライアントがそのまま使えます。
ElastiCache Serverless for Valkeyは、このValkeyをAWSがフルマネージドで提供するサービスです。
ノードのサイズや台数を事前に見積もる必要がなく、アクセス量に応じて自動でスケールする「Serverless」形態を選ぶことができるという特徴があります。
2026年5月にメジャーアップデート(9.0)が行われ、GEOSEARCHにポリゴン範囲を指定できるBYPOLYGONオプションなど、地理空間検索まわりの機能も強化されました。（今回は未使用）
インデックス化
位置情報検索と最寄り情報の検索にあたって、以下のインデックスを登録しました。
-
GEOADD：緯度経度を空間インデックスとして登録
→GEOSEARCHでO(log N)の最寄り検索 -
ZADD：値をスコア（今回は時刻）付きのソート済み集合として登録
→ZRANGEBYSCOREで「この時刻以降」を一発取得
実装
事前にインデックスを作る「登録」と、リクエストのたびに呼び出す「検索」の2段階に分けて実装を行いました。
python
# 登録（事前準備）
r.geoadd("bus_stops", (lon, lat, stop_id))
r.zadd(f"departures:{stop_id}", {f"{dep_time}|{trip_id}": score})
# 検索
r.geosearch("bus_stops", longitude=lon, latitude=lat, radius=1000, unit="m", sort="ASC", count=1)
r.zrangebyscore(f"departures:{stop_id}", after_score, "+inf", start=0, num=5)
データ投入で発生したClusterCrossSlotError解消方法
データ投入時、以下のエラーに遭遇しました。
ClusterCrossSlotError: Keys in request don't hash to the same slot
ElastiCache Serverlessは裏側でクラスターモードとして動いており、pipelineでまとめて送れるのは同じスロットに属するキー同士だけという制約があります。GEOADDとZADDを別々のキーに対して大量にパイプライン送信していたため、このエラーが発生しました。
対処法は2つです。
1. クラスター対応のクライアント（redis.cluster.RedisCluster）に切り替える
2. 一定件数（今回は2,000件）ごとにpipe.execute()し、こまめに送信する
python
from redis.cluster import RedisCluster, ClusterNode
r = RedisCluster(startup_nodes=[ClusterNode(host, port)], decode_responses=True, ssl=True)
pipe = r.pipeline()
for i, row in enumerate(rows):
pipe.geoadd(...)
if i % 2000 == 0:
pipe.execute()
pipe = r.pipeline()
pipe.execute()
実測結果
インデックス化後の実測結果は以下の通りです。
Lambda（arm64/Graviton2）からElastiCache Serverless for Valkeyへ問い合わせる構成で、計測はCloudWatch Logsで行いました。
| 実行回 | GEOSEARCH | ZRANGEBYSCORE | 合計 | 備考 |
|---|---|---|---|---|
| 1回目 | 272.73ms | 1.39ms | 274.12ms | コールドスタート |
| 2回目 | 35.56ms | 1.32ms | 36.88ms | ウォームアップ中 |
| 3回目 | 7.13ms | 1.13ms | 8.27ms | ほぼウォーム状態 |
コールドスタートの影響で1回目は大きく出ていますが、ウォーム状態では合計8.27ms、うちZRANGEBYSCOREは1.13msまで縮んでいます。
743.8msだった愚直実装と比べて、約90倍高速化することができました。
まとめ
ElastiCache Serverless for Valkeyでインデックス化することで、データ検索を高速化することができました。
位置情報検索やソート済みデータで全件検索・多段検索を行う場合などにぜひ活用してみてください。