Blitz（後編）：リアルタイムユーザー解析エンジンを実現する技術 - 強整合な解析 - 

<html><head></head><body><h2 id="Introduction">Introduction</h2>
PLAIDでは9年ほど前から、大規模ストリーミングデータから高速に、Aggregation/Filteringをするモジュールを育ててきて、KARTEというプロダクトの基盤としてきています。Personalizeのようなタスクをツールにガシガシ組み込むために必要なモジュールで、かなりニッチな情報ということもあって外部にあまり発信する機会がありませんでした。（決して面倒だったからではありません。決して。。）
ただ、少し前にそのエンジンの刷新プロジェクトが終わったところで、そのプロジェクトの発信と併せて、初期コンセプト上の面白み（コード上は2度3度と刷新されていますが、初期コンセプトの面白い部分は踏襲されつつ新しいものが加わっています）もそれなりにあるし、発信してみようと思って書き始めました。
<h2 id="要求されること">要求されること</h2>
現在は コードネーム “Blitz” と呼ばれる分析モジュールは、Personalizationに特化したもので、スキーマが固定されないJson形式での次のようなイベントストリーム
<pre><code class="language-jsx hljs">{
 $meta: {
 name: "nashibao",
 isMember: 1
 },
 $buy: {
 items: [{
			sku: "xxx",
			price: 1000,
 }]
 }
}
</code></pre>
[図1: イベントデータ]
がuserIdに紐づいて飛んできた時に、次のようなフィルタリングを高速に処理することができるものです。
<pre><code class="language-jsx hljs">match("userId-xxx",
 DAY.Current('$meta.isMember', 'last') = 1,
 ALL.Current('$buy.items.price', 'avg') &gt;= 10000,
 WEEK.Previous('$buy.items.price', 'sum') &lt;= 100,
 WEEK.Previous('$session', 'count') &gt; 10,
 ...
)
</code></pre>
[図2: ルール定義の疑似コード]
これは “userId-xxx” が “会員” であり、1年間では購買単価が1万円以上で、先週のセッション数が10以上に関わらずほとんど買い物に繋がらなかったユーザー なのかどうかを検証するルールを示した疑似コードです。
この時に、厳しい性能要件をクリアするように作られています。
<div class="TableScroll"><table>
<thead>
<tr>
<th>種類</th>
<th>説明</th>
</tr>
</thead>
<tbody>
<tr>
<td>スケーラビリティ</td>
<td>- High Trafficなユーザー数やイベント数に対してスケールする - 時間経過と共に性能が劣化しない - 時間経過と共にストレージコストが爆発しない</td>
</tr>
<tr>
<td>低レイテンシ</td>
<td>- 数百ms以内にレスポンスする</td>
</tr>
<tr>
<td>新鮮性</td>
<td>- 計算にラグが発生しない = 数秒前のイベントも考慮される （初期ver: 結果整合性 → 現在: 強整合性)</td>
</tr>
<tr>
<td>スキーマレス</td>
<td>- 送信されるイベントは、スキーマが固定されない - 多様なデータ型（Nest / Array含む）が送信される</td>
</tr>
<tr>
<td>即時更新性</td>
<td>- 検証ルールの変更は、数分でデプロイされる - 過去データを遡って考慮する</td>
</tr>
<tr>
<td>自由度</td>
<td>- 多種多様なタイムウィンドウ、演算子に対応している</td>
</tr>
</tbody>
</table></div>
特にルールやイベントの自由度を高く維持する割に、過去データの再集計などでラグが発生せずにルールの変更を反映し、数百ms以内にレスポンスを返せる状態に持っていくことが非常に <s>めんどくさい</s> 難しい要件になります。
既存の基盤やフレームワークをカスタムして作る方針で初期は検討しましたが、なかなか要件に見合わなかったため、ストレージレイヤーより上部に関しては主に自社で構築しています。
<div class="TableScroll"><table>
<thead>
<tr>
<th>Name</th>
<th>スケーラビリティ</th>
<th>低レイテンシ</th>
<th>新鮮性</th>
<th>スキーマレス</th>
<th>即時更新性</th>
<th>自由度</th>
</tr>
</thead>
<tbody>
<tr>
<td>Embed (単純に事前に処理を組み込んだもの)</td>
<td>◉</td>
<td>◉</td>
<td>◉</td>
<td>x</td>
<td>◉</td>
<td>x</td>
</tr>
<tr>
<td>Batch Processing (ex: MapReduce)</td>
<td>◉</td>
<td>x</td>
<td>x</td>
<td>◉</td>
<td>-</td>
<td>◉</td>
</tr>
<tr>
<td>Stream Processing (ex: Beam)</td>
<td>- (過去データは捨てられる)</td>
<td>◉</td>
<td>◉</td>
<td>◉</td>
<td>x</td>
<td>◉</td>
</tr>
<tr>
<td>Lambda Architecture</td>
<td>◉</td>
<td>▲ (x 1sec未満)</td>
<td>◉</td>
<td>◉</td>
<td>x</td>
<td>◉</td>
</tr>
<tr>
<td>Blitz</td>
<td>◉</td>
<td>◉</td>
<td>◉</td>
<td>◉</td>
<td>◉</td>
<td>▲</td>
</tr>
</tbody>
</table></div>
SQLほどの自由度は持たせない代わりに、他の要件を満たすように作られています。考え方としては、Aggregation処理を事前処理と事後処理の2つの処理に分解しているという意味で、Lambda Architectureや一部のOLAPに非常に近く、それらをStream処理向けに実装したものと理解するのが近いかもしれません。
<h2 id="ストレージ&amp;計算コスト">ストレージ&amp;計算コスト</h2>
<pre><code class="language-jsx hljs">Space Complexity ∝ Time Complexity = O(T)
</code></pre>
通常、ルールを変更した上で過去データも含めて計算し直すためには、時系列データをスキャンする必要があるため、必要なストレージは経過時間(<code>T</code>)に比例して増えていきます。また処理によっては分散も可能ですが、計算コストもそのストレージサイズに比例して増えるので、結果的に時間に比例して増えていきます。
これに対してBlitzでは
<pre><code class="language-jsx hljs">Space Complexity ∝ Time Complexity = O(D)
</code></pre>
時系列でのデータを、後続処理での自由度を減らさないように送信イベントのスキーマのフィールド数(<code>D</code>)ごとに最小単位の統計値の集合に圧縮し（Pre-Aggregation処理）（この最小単位の統計値の計算の種類は実は大きいですが、固定値）、これらを検証ルールへの適合時にもう一度組み合わせる (Post-Aggregation処理) ことで、ストレージコスト、計算コスト共に、時間(<code>T</code>)ではなくイベントスキーマのフィールド数(<code>D</code>)に比例するように変更しています。
<img src="https://ik.imagekit.io/plaid/ba74a798-8f73-4a03-8aa6-8199e7c74bf8/image.png" alt="image.png">
様々なタイムウィンドウと演算子を念頭に置き、多様な統計量を用意するために、事前にある程度大きな空間をAggregate軸（この場合ユーザー)ごとに用意して置く必要があります。そのため経過時間に対して、実は初期は過大に空間を食いますが、経過時間と共に処理後の方が空間・計算コスト共に効率的になっていく想定です。
<img src="https://ik.imagekit.io/plaid/e56ce33a-4ad3-423d-b408-4d41e655955d/image(1).png" alt="image(1).png">
<h2 id="Pre-Aggregation処理">Pre-Aggregation処理</h2>
処理のフローやアーキテクチャの考え方には面白いところがたくさんありますが、ここではストレージサイズに話を絞ることにし、Pre-Aggregation処理によるデータ変換をどのように行なっているかを簡単に紹介します。
統計的な圧縮処理（Pre-Aggregation）のデータフローの簡易的な図は以下です。BlitzではMapReduceのReduce処理と同じように、Tree構造で分解可能な処理のみを許しています。そうすることで処理の分散性を保つと共に、下に示すような統計値同士のRollupも可能にするため、さらにストレージサイズを下げることができます（たとえば、日のデータを7つ持っていれば週のデータを作れます）。
<img src="https://ik.imagekit.io/plaid/af418988-bcb7-4dfa-a4d7-951c9bfd8d0e/image(2).png" alt="image(2).png">
Aggregationの自由度や新鮮性を達成するために、Reduce処理の実態は非常に複雑で、CAS操作のようにEventそのものと様々なTimewindowの統計値を読み出し結合した上で、上書き(Swap)する操作を含んでいます。
どのようにRollupを計画するのか、競合をどのように排除するかも非常に面白い部分ですが、ここでは紹介しません。
このようにすることでRead時に消費したEventや古くなったSlotはどんどん破棄していくことができます。
<img src="https://ik.imagekit.io/plaid/e4456103-e9bb-4305-906a-d9f64810dac8/image(3).png" alt="image(3).png">
<h2 id="最後に">最後に</h2>
さて前編はここまでにします。紹介した通り、この初期コンセプトの考え方は至ってシンプルなものです。しかし、机上ではシンプルですが、実際の実装上においては、Hight Trafficを裁くための技術やHot Spotを避けるための技術、強整合を保つための技術などなど、様々な面白さが詰まっています。
<a href="https://tech.plaid.co.jp/blitz_realtime_analytics_architecture">後編</a>ではBlitzの開発メンバーがその実装に入り込んだ面白さを紹介します。
</body></html>

database, analytics, karte, algorithm, aggregation, blitz

Blitz（前編）: 自由度と即時更新性を担保したAggregation