Introduction

KARTEを支えるリアルタイムユーザー解析エンジン（Blitz）のパフォーマンスに関する記事です。解析基盤が抱えていたボトルネックを解消した方法について、新旧のアーキテクチャの違いにフォーカスしながら説明していきます。

以下記事の続編にあたるので、事前に読むのがオススメです。
Blitz（前編）：自由度と即時更新性を担保したAggregation
Blitz（後編）：リアルタイムユーザー解析エンジンを実現する技術（強整合な解析）

課題

KARTEのエンドユーザー(以下ユーザー)の統計情報は平均でも100kb以上と大きく、取得コストはやや大きいです。統計情報が大きくなる要因として、以下が挙げられます。

• 統計値をフィールドやタイムウインドウごとに複数種類保持
• 一部の統計値では送信された文字列をそのまま保持

下記は統計情報のイメージです。

{
  $week: {
    $buy: {
      sku: {
        last: "xxxxxxxxxxxxxxx",
        first: "yyyyyyyyyyyyyyyyyy",
        sets: ["xxxxxxxxxxxxxxx","yyyyyyyyyyyyyyyyyy"],
        ....
      },
      price: {
        last: 100,
        sum: 2000,
        max: 200,
        ...
      }
    }
  }
}

統計情報はBigtableに保存しています。

旧アーキテクチャでは、ユーザーのイベントが発生するたびにBigtableからこのユーザー統計情報を取得しており、全体の中で最もコストとレイテンシがかかる処理でした。

キャッシュの導入は整合性の問題で困難でした。Bigtableから取得した統計情報をそのままキャッシュすると、一部のイベント(まだRealttime Queue上にあってConsumeされてないイベント)が畳み込まれていない統計情報を使い続けることになってしまうからです。

このような要因からアクション配信のレイテンシはp95で1.5秒程度でした。ユーザー体験を損ねているというフィードバックをお客様からいただくこともよくありました。

このボトルネックをBlitzで解消した方法について説明します。

1. 取得するデータの絞りこみ

Blitzのバックエンドは、アクションを配信するサーバ(以下アクションサーバ)と、統計情報を永続化するサーバに大きく分かれます。
アクションサーバでは、ユーザーの統計情報を使ってパーソナライズされたアクション(メールやポップアップ)を配信します。

ここで、保存されてる統計情報のうち実際に配信に使われる統計値はごくわずかです。
旧アーキテクチャではほとんど全ての統計情報を取得していましたが、Blitzでは、お客様が設定したアクションの配信設定から必要最小限の統計情報のフィールドとBigtableのフィルタを算出し、データベースから取り出しています。

一見シンプルに見えますが、以下の点が課題でした。

• 配信設定に関する機能は多岐に渡り、必要な統計値を算出するための対応箇所が多い
• 統計値間の依存関係がある
  • 例：「平均」は「合計と回数」に依存(後述の通り統計情報をキャッシュ上で更新し続けるため、依存する統計値も必要)
• 統計値のタイムウインドウ間の依存がある
  • 例：「前回のセッション」は「最新のセッション」に依存(こちらもキャッシュ上で更新し続けるために必要)
• Bigtableのフィルタは細かく設定しすぎるとCPU消費が大きくなり逆効果
  • Bigtableのフィルタだけだと絞り込みきれないから、デシリアライズ時にも余分なフィールドの実体化を省略したい

対応として、フィルタの中間表現を定義し、統計値間の依存を考慮して中間表現を算出するモジュールを作ることで、安全かつ簡単に統計値への依存を追加できるようにしました。また、中間表現を挟むことで、Bigtableとデシリアライザのフィルタリングもシンプルに実装できました。この中間表現はMongoDBの機能になぞらえてProjectionと呼んでいます。

こちらは全体のフローのイメージです。

またデータフォーマット自体もJSONからMessagePackに変更して、少量化とSerialize/Deserializeの改善をしました。

(厳密にはこれらの最適化自体はリアーキテクチャせずとも実装可能でしたが、改修の難易度が高かったので、リアーキテクチャに伴うコードベースとデータの刷新に混ぜ込む形で実装しました)

2. キャッシュ

絞り込みによって一定の改善ができたものの、依然として取得処理は重いです。
更なる改善として、Blitzではユーザーの統計情報をアクションサーバのメモリにキャッシュしています。

キャッシュの整合性と効率に焦点を当てつつ、その仕組みについて説明します。

前記事でも述べた通り、Blitzではアクションサーバの前段にBigtableベースのイベントキューがあります。また、キューにイベントが追加されるとLoadBalancer経由でアクションサーバに通知されます。

どのアクションサーバに通知が届いても、そのサーバは、今までに送られた全てのイベントが畳み込まれた最新のユーザー統計情報を扱わないといけません。一見難しく見えますが、強整合性が担保されたキューが前段にあることで、シンプルに実現できました。

キューのチェックポイントも統計情報と一緒に保持し、通知を受けたらそのチェックポイント以降のデータをキューからスキャンすればいいのです。

以下は、アクションサーバの処理フローを示した疑似コードです。

handle_notification(user_id):
    user = cache.get(user_id)
    if user is None:
        user = db.get_user(user_id)
    
    # チェックポイント以降のイベントをBigtableからスキャン
    events = scan_events(user_id, user.checkpoint)
    
    for event in events:
        new_user = reduce(user, event)
        deliver_personalized_action(event, new_user)
    
    new_user.checkpoint = events[-1].timestamp
    
    cache.put(user_id, new_user)

基本的なロジックはこれだけです。

また、LoadBalancerにはGCPのL7 Internal LoadBalancerを採用しました。Session Affinityのキーにuser_idを指定して同じユーザーのイベントをできるだけ同じサーバに振り分けることで、キャッシュ効率を上げています。Consistent Hashを使うことで、バックエンドの台数が増減してもキャッシュミスは少なく収まります。

データをしっかり絞り込んだおかげか、メモリやGC関連で苦戦を強いられることもこれといってありませんでした。machine typeも特別メモリが大きいものではないです。

ここまでの説明だと簡単そうに見えますが、実際には以下のような点について、さらに複雑なロジックを実装しています。

• アノニマスユーザーがサービスにログインしたときに、サービスのユーザーIDでマージ
• イベントの順序性の担保
  • ナイーブにBigtableのキューのキーにタイムスタンプを付与するだけだと、クロックのズレなどの要因で順序担保ができません
• Statsを都度作り直すと重いので、スレッド間の競合制御をしながらキャッシュをmutableなobjectとして更新

結果

他にも様々な改善が加えられた新アーキテクチャに移行して、以下が達成できました。

• Compute Engineのコストが7割減
• アクション配信のレイテンシ(p95)が1500msec -> 200msec
  • 配信が遅いという声も聞かなくなりました
• Bigtableのコストが数割減(正確なデータが探せませんでした…)

裏側としてはなかなかいい成果だったと思います。ただ、2022年の円安が加速している時期に移行してコスト削減の効果が相殺されたことなどもあり、最終的には少し派手さに欠ける結果になったのを覚えています笑

まとめ

以前のアーキテクチャでは解析が非同期であるがゆえに、結果整合性しか担保できず、統計情報のキャッシュもできていませんでした。

BlitzではBigtableのキューを前段に置くことで、整合性とパフォーマンスの2つの問題が解決できました。さらに言えば、受信したリクエストをすぐにキューに積むため、データが欠損するリスクへの耐性も上がったと言えます。

1つのアイデアで複数の課題が解けたところは面白い点だと思います。

他にも様々な最適化や運用上の工夫が凝らされているので、今後紹介できればと思います。