ランチメンバーを自動でグループ分けしてくれるSlackbotをAWS Lambdaでサクッと作ってみた

<html><head></head><body>プレイドの @nashibao です。
弊社では結構<code>BigQuery</code>を使い倒させていただいていて、社内向けのバッチ解析や<code>ChartIO</code>/<code>ModeAnalytics</code>等を介した社外向けのレポーティングとしての利用だけでなく、ABテストなどの集計系では（Query数のコントロールのためにキャッシュは介していますが）ほぼ直接アプリケーションのバックエンドとして利用しています。
またStreaming Insertが思ったより安定しているので、ニアリアルタイムにイベントを反映することができ、適当なQueryを投げ込んでPBの集計を気軽にやって、数百万使って唖然としてとりあえず寝て忘れる、みたいなことをしています。
先日も"GCP NEXT World Tour in Tokyo"において、弊社 @makinoy がジョブズばりのプレゼンをさせてもらいました。
<a href="http://blog-plaid.com/event/2642/" target="_blank" rel="noreferrer"><img src="https://ik.imagekit.io/newt/61b151f921640c0018173598/df784d2b-d07b-4eac-b2d1-afb1e918b464/DSC00529.jpg" alt="gcp"></a> 
<a href="http://blog-plaid.com/event/2642/" target="_blank" rel="noreferrer">Google Cloud Platform 国内最大のイベント「GCP NEXT World Tour in Tokyo」にプレイドが登壇しました</a>
さて、これだけBigQueryを使い込んでいる弊社ですが、仕組みについてはあまり外に出てないため、ほとんど知りません。
<a href="http://research.google.com/pubs/pub36632.html" target="_blank" rel="noreferrer">DremelのPaper（2010/VLDB)</a> が6年前に出ていますが、その内容としては非常に単純なQuery Processingのケースのみを扱っていて、<code>シンプルな分割統治型のツリー処理</code> と、<code>独自の省スペースかつ効率的なNested Dataのカラムナ表現</code> の二点を中心に議論されています。論文の最後でも次のように締められています。
<blockquote>
In the future, we plan to cover in more depth such areas as 
formal algebraic specification, joins, extensibility mechanisms, etc.
</blockquote>
後半に乗っている<code>Dremel</code>と既存の<code>Google MapReduce/Sawzall</code>（Google 社内のMapReduce実装）とのパフォーマンス比較、キレイに線形にスケールする実験結果は一見の価値があります。ただ、カラムナについて独自のエンコーディング/FBMを使ったアセンブリ手法は、これがポッと作れてしまうGoogleはさすがだなぁ、と思う反面、カラムナにしたらそりゃパフォーマンスは上がるだろうという感じだし、割となんでもござれのHDDベースのMapReduceと比較して分割統治に特化した構造がパフォーマンスが出るのも当然だろ、という気がします。
それよりもむしろ、<code>テラ/ペタサイズのJOINやGroup Byが入り組んだクエリが数秒~数十秒で返ってくる</code>という理解不能な事象について深掘りして欲しい気がします。つまりちょうど<code>Dremel</code>のペーパーの範囲外ですw
<a href="https://www.amazon.co.jp/Google-BigQuery-Analytics-Jordan-Tigani-ebook/dp/B00JUUZIZI" target="_blank" rel="noreferrer">Amazon.co.jp: Google BigQuery Analytics 電子書籍: Jordan Tigani …</a>
<img src="https://ik.imagekit.io/newt/61b151f921640c0018173598/5438d3eb-fff3-4106-bd6e-4db25c4b588e/583ec39bdd099caa2040f48c::image.png" alt="img">
そしてようやく本題ですが、JOINを含めたQuery Processingについては2014年に出ているこちらの本の方が詳しいので、少し紹介したいと思います。 
ちなみにタイトルは（かなり）盛り気味です。
<h1 id="先に感想">先に感想</h1>
いくつかポイントがあると思いますが、僕が個人的に思うBigQueryのキモは次の3点です。（ちょっと想像で物を言っているので間違ってたら、、すみませんw）
<ol>
<li>まずまずDisk IOの並列化（並みの並列化じゃない）</li>
<li>スピード重視でメモリとネットワークを酷使する設計（並みの酷使っぷりじゃない）</li>
<li>Slowなところを避ける代わりに出来なきゃ死ぬ設計。やり直せばいいという割り切り（その点逆にMapReduceはすごい）</li>
</ol>
Disk IOを並列化して、読込をボトルネックから解放するのはこれ系では前提感あります。それに加えてネットワークを酷使したり、Hash PartitionによるShufflingなどで、メモリオーバーでabortするリスクをとるポリシーあたりがHadoopを含めたMap Reduce界隈と違うのかな、と感じます。
<h1 id="基本構造とベーシックなクエリ処理">基本構造とベーシックなクエリ処理</h1>
基本的な構造としては次のようになっているようです。
<img src="https://ik.imagekit.io/newt/tr:w-1000,h-1000,c-at_max/61b151f921640c0018173598/55a1103a-d5c9-4f85-a446-1a0f8faa8ea2/583ec39bdd099caa2040f48b.png" alt="583ec39bdd099caa2040f48b.png">
クエリは<code>Root/Mixer/Slot</code>の順で伝播します。その時Slotのみで解決できないQuery要素に関しては排除されて伝播されます。<code>Slot</code>は実際にStorageレイヤと通信し最初の処理をする<code>Leaf Server</code>のスレッドのことのようです。そして、処理結果がSlot/Mixer/Rootの順で伝播し、それぞれ統治処理が行われます。
例えばJOINやGroup Byが絡まず、単純な分割統治処理のようなケースの場合
<pre><code class="hljs">SELECT corpus, word, word_count FROM tbl 
 WHERE LENGTH(word) &gt; 4 
 ORDER BY word_count 
 DESC LIMIT 5
</code></pre>
（word_countが最初からaggregationされているという現実にはなさそうなケースですが）単純に同じQueryがSlotに伝播され、それぞれのSlotが返す結果に対してMixerが同じQueryをかければ終了です。
<h1 id="Small Group By">Small Group By</h1>
次のようにGroup Byがある場合も分割統治型で処理されます。
<pre><code class="hljs">SELECT corpus, word, SUM(word_count) AS total FROM tbl 
 WHERE ... 
 Group By corpus 
 ORDER BY total 
 DESC LIMIT 5
</code></pre>
まず <code>Order</code>句と<code>LIMIT</code>句を排除してSlotに渡しクエリし、その結果をMixerに渡し、Mixerでは元のクエリをかけて終了です。ただしDistinctなKeyが多い場合メモリに乗り切らずabortします。
<h1 id="BroadCast Join">BroadCast Join</h1>
さてJoinですが、単純な場合としてJoinする片方がSlotにバラまけれるサイズであれば分割統治型の繰り返しで対応できます。（BigQueryでは <code>8mb</code>が閾値のようです）。こちらは
<pre><code class="hljs">SELECT hoge FROM a 
 JOIN 
 (SELECT fuga FROM b) AS c 
 ON a.hoge = c.fuga
</code></pre>
<ol>
<li>サブクエリ<code>SELECT fugue FROM b</code>のみをSlotに配布</li>
<li>Mixerに結果が返される</li>
<li>返された結果をインライン展開して、全体のクエリを配布</li>
<li>Mixerに結果が集められ、Aggregationして終了</li>
</ol>
<img src="https://ik.imagekit.io/newt/61b151f921640c0018173598/0f92e382-bd5f-4629-b49c-fa7a9c372f90/583ec39bdd099caa2040f48a.png" alt="583ec39bdd099caa2040f48a.png">
一つ面白いのは、例えば5000 Leafあれば 5000 x 8MB = 40GBのネットワークコストをかけている、というところです。
<h1 id="Shuffled Query（Big Join/Big Group By）">Shuffled Query（Big Join/Big Group By）</h1>
最後に複雑なケースとして、大規模なJOINと大規模なGroup Byのケースです。MapReduceと同じようにShuffleフェーズを組み合わせて処理をしますが、Merge Sortではなく、単純なHash Partitioningで分散します。つまり100台のLeafで処理していれば、keyに対して100でmodを取り対応する各Leafに配布します。
（元々は <code>JOIN EACH</code>や<code>GROUP EACH BY</code>といった特殊な句が用意されていましたが、自動で判断するようになったようなので、省きます。( <a href="https://cloud.google.com/bigquery/docs/reference/legacy-sql" target="_blank" rel="noreferrer">Query Reference </a> )）
<pre><code class="hljs">SELECT hoge FROM a 
 JOIN 
 (SELECT fuga FROM b GROUP BY fuga) AS c 
 ON a.hoge = c.fuga
</code></pre>
<ol>
<li><code>b</code>に関するサブクエリをLeafに配布する。（例えば100Leaf）</li>
<li><code>fuga</code>を100個にHash Partitioning（<code>shuffle</code>）し、Leaf間でデータを交換する</li>
<li>Leafでサブクエリを実行しMixerに終了したかどうかだけを返す（データは返さない）</li>
<li>次に<code>a</code>に関するクエリをLeafに配布する（例えば500Leaf）</li>
<li><code>hoge</code>を500個にPartitionしLeaf間でデータを交換する</li>
<li>さらに3のデータも500個にPartitionし、Leaf間でデータを交換する（この処理がどこに挟まるかちょっと書いてなくて想像で書いてます）</li>
<li>Shard上でJOINのクエリを実行し、Mixerに結果を返す</li>
<li>Mixerで統合し終了</li>
</ol>
<img src="https://ik.imagekit.io/newt/61b151f921640c0018173598/6e7a1ef9-4119-4d18-8b8c-dba18a848b52/583ec39bdd099caa2040f489.png" alt="583ec39bdd099caa2040f489.png">
大きな注意としては、どでかいデータでHashが偏ると乗り切らずAbortする、という点です。
<h1 id="最後に">最後に</h1>
先に書いた感想と同じですが、割と古典的というか泥臭いやり方でShuffleを実装したりしているところ、ネットワーク/メモリを酷使し、合わないものはAbortするところなど、失敗を許容する代わりにパフォーマンスを出す設計が重要なように思います。
もう少しMapReduce界隈との設計思想の違いについても想いを馳せてみたかったんですが、図を作っていたら力尽きた感じがあるので、議論したい方は是非ともPlaidの社内勉強会等に来てください！
ウェブ接客プラットフォーム「KARTE」を運営するプレイドでは、 KARTEを支える技術に興味を持つエンジニア（インターンも！）を募集しています。 
詳しくは<a href="https://plaid.co.jp/recruit.html" target="_blank" rel="noreferrer">プレイドの採用ページ</a>か、<a href="https://www.wantedly.com/companies/plaid/projects" target="_blank" rel="noreferrer">Wantedly</a>をご覧ください。
</body></html>

プレイドで使い倒しているBigqueryの内部構造について徹底解剖してみました！

Bigqueryの内部処理について徹底解剖してみた