Uberがリアルタイムマーケットプラットフォームをスケールしている方法

Uberは、たった4年で38倍という目覚ましい成長を遂げたという。今回が恐らく初めてだと思うが、UberのチーフシステムアーキテクトであるMatt Ranney氏が、その面白くて詳細にわたる発表、「Scaling Uber’s Real-time Market Platform（Uberのリアルタイムマーケットプラットフォームをスケーリングする）」の中で、Uberのソフトウェアの仕組みについて詳しく説明している。

もしあなたが「急騰料金（サージ・プライシング）」に興味があるなら、それはこの発表の中に出てこない。知ることができるのは、Uberのディスパッチシステム、同社の地理空間インデックスの実装方法、システムのスケール方法、高可用性の実現方法、不具合の対処法についてだ。例えば、ドライバーの電話をリカバリ用の外部分散ストレージシステムとして使うという、データセンターの不具合に対処する驚きの方法などについても知ることができる。

講義の全体的な印象は、かなりの急成長をしているといったところだ。急激な成長に伴い、できたばかりのチームが可能な限り迅速に動けるように権限を与えようとした結果は、同社がおこなってきたアーキテクチャにおける選択に影響を与えている。これまでの同社の大きな目標は、チームのためにエンジニアリング速度をできる限り高めることであった。そのため、技術の大部分がバックエンドに利用されてきた。

当然ながら雑然とした（そして非常に首尾よい）スタートを切った後、Uberは自社のビジネスや、成功に本当に必要なものについて、多くを学んだようだ。同社の初期のディスパッチシステムは、典型的なもので、人物だけの移動を深く想定して作られていた。今やUberのミッションは、人だけでなく、宅配便や食品に対処することにまで成長した。そのため、同社のディスパッチシステムは抽象化され、非常に強固でスマートなアーキテクチャ基盤となった。

Matt氏は、アーキテクチャが少々クレイジーかもしれないと考えているものの、コンシステントハッシュリングとゴシッププロトコルを使用するというアイデアは、同社のユースケースにピッタリだ。

Matt氏の、取り組みに対する溢れんばかりの情熱には、魅了されずにはいられない。同社のディスクサブシステムであるDISCOは、学校で習う巡回セールスマン問題みたいだと、Matt氏は興奮した口調で話した。これは、まさに素晴らしいコンピュータ・サイエンスだ。ソリューションは最適ではないものの、フォールト・トレラントで拡張可能な要素から構成されていて、面白い規模におけるリアルタイムかつ現実世界での巡回セールスマンなのだ。なんて素晴らしいんだろう！

さて、Uber内部での取り組みを見てみよう。Matt氏の発表を僕なりに解釈したものを以下にまとめた。

状況

• Uberの地理空間インデックスにおける目標は、毎秒100万書き込み。読み込みはその数倍。
• ディスパッチシステムには何千ものノードがある。

プラットフォーム

• Node.js
• Python
• Java
• Go
• iOSとAndroidのネイティブ・アプリケーション
• Microservices
• Redis
• Postgres
• MySQL
• Riak
• Twitterの Twemproxy (Redis用)
• GoogleのS2ジオメトリ ライブラリ
• ringpop ：コンシステントハッシュリング
• TChannel ：ネットワーク多重化およびRPC用フレーミングプロトコル
• Thrift

概要

• Uberは、乗客とドライバーを結び付ける輸送プラットフォームだ。
• 課題： 動的需要と動的供給をリアルタイムにマッチングすること。供給側のドライバーは、自由にやりたいことができる。需要側の乗客は、好きな時に輸送を要することができる。
• Uberのディスパッチシステムは、携帯電話を利用してドライバーと乗客をマッチングするリアルタイムマーケットプラットフォームである。
• Uberにとって、大晦日が最も忙しい時である。
• 業界がいかに早く、こんなにも大きな前進を遂げてきたかということを人々は忘れやすい。技術はますます良く、速くなってきていて、つい最近まで素晴らしいと思っていたものでも、あっという間に影が薄くなってしまう。2、30年前は携帯電話、インターネット、GPSはほぼSFレベルだったが、現在僕たちがそう意識することはほとんどない。

アーキテクチャ要旨

• 全てを動かしているのは、ネイティブ・アプリケーションを携帯電話上で実行している乗客とドライバーだ。
• バックエンドは主に、携帯電話のトラフィックにサービスを提供している。クライアントはモバイルデータ通信やベスト・エフォート型インターネットを通じてバックエンドと接続する。モバイルデータ通信に基づくビジネスを10年前に想像できただろうか？現在このようなことをできるのは素晴らしいと思う。使われるのは、プライベートネットワークでも、手が込んだQoS (Quality of Service)でもなく、オープンなインターネットだけだ。
• クライアントは、ドライバーと乗客、つまり需要と供給をマッチングするディスパッチシステムに接続する。
• ディスパッチはほぼ完全にnode.jsで書かれている。
  • jsに移す予定だったが、その後io.jsとnode.jsが統合した。
  • javascriptで動く、面白い分散システムを構築することができる。
  • 熱意が生む生産力を過小評価するのは難しく、node開発者は非常に熱心である。彼らはとても素早くたくさんの仕事をこなす。
• Uberのシステム全体は非常にシンプルだと思われる。なぜこれら全てのサブシステムとこれだけ多くの従業員が必要なのだろうか？そう見えるなら、それは成功のしるしだ。やるべきことはたくさんあるし、システムがシンプルに見えているということは、サブシステムや従業員が仕事を全うしたということだ。
• 地図／ETA（到着予定時刻）。ディスパッチが賢い選択をするには、地図と経路情報を取得する必要がある。
  • ストリートマップと過去の移動時間が現在の移動時間の見積もりに使われる。
  • 言語は、どのシステムが統合されているかに大いに依存する。そのため、Python、C++、Javaがある。
• サービス。膨大な量のビジネスロジックサービスがある。
  • マイクロサービスのアプローチが使われる。
  • 主にPythonで書かれている。
• データベース。様々なデータベースが使われる。
  • 最も古いシステムはPostgresで書かれていた。
  • Redisが多く使われている。Twemporoxyで使われることもあれば、カスタムクラスタリング・システムで使われることもある。
  • MySQL
  • Uberは、たくさんのMySQLインスタンスをまとめる独自の分散カラムストアを構築している。
  • ディスパッチサービスには、Riakに状態を保存しているものもある。
• 輸送後のパイプライン。多くの処理は輸送が完了した後に発生しているはずだ。
  • レーティングを集める。
  • Emailを送る。
  • データベースを更新する。
  • 支払いを計画する
  • Pythonで書かれている。
• お金。Uberには多くの支払いシステムが統合されている。

旧ディスパッチシステム

• 元のディスパッチシステムにあった制限は、会社の成長を抑制し始めていたため、変える必要があった。
• Joel Spolskyが何と言おうと、ほぼ全部書き換えられた。その他のシステムは全く触れられておらず、ディスパッチシステム内のサービスでも書き換えられずに残ったものもある。
• 旧システムは個人輸送のために作られたが、それは多くの憶測を生むことになった。
  • Uber Poolを実現できない、車一台につき一人のの乗客
  • 人物移動のアイデアに際し、データモデルとインターフェイスに至るまで深く考えられた。このことで、新市場や新製品（食料や配達物の移動など）への参入を制約することとなった。
  • 元のバージョンは都市に分散された。そうすることで各都市が別々に運営できるので、スケーラビリティのためには良い。追加される都市が増え、次第に管理できなくなってきた。大都市もあれば、小さい町もある。負荷が大きく急上昇する都市もあれば、そうでない町もある。
• 急速に構築されているものがたくさんあるため、同社は単一障害点でなく、多重の障害点を抱えている。

新ディスパッチシステム

• 都市分散を調整し、より多くの製品をサポートするために、需要と供給の考えを汎用化する必要があった。そのため、サプライサービスとデマンドサービスが作り出された。
• サプライサービスは、供給全体の機能性とステート・マシンをトラッキングする。
  • 車をトラッキングするために、モデルには多くの属性が用意された。シートの数、車種、チャイルドシートの有無、車椅子を乗せられるかなど。
  • 割り当てをトラッキングする必要がある。例えば、車一台に3つ座席があっても、そのうち2席は使用中かもしれない。
• デマンドサービスは要求、注文、そして需要のあらゆる側面をトラッキングする。
  • 乗客がチャイルドシートを要求した場合、その要求を在庫に照合させなければならない。
  • もし乗客が、料金が安くなる代わりに一台に相乗りしても良いという場合、モデル化されなければならない。
  • 宅配物を移動させる必要がある場合や、食料が配達される場合は？
• 需要と供給全てをマッチングするロジックは、DISCO（派遣最適化）と呼ばれるサービスである。
  • 旧システムでは、現在有効な供給、つまり、現在道で待機中の車のみをマッチングしていた。
  • DISCOは、先のことを計画し、車が輸送可能になった際の情報を活用するサポートをしてくれる。例えば、輸送中のルートの見直しなど。
  • geo by supply（供給に基づく地理）。全ての供給がある場所やそれが想定される場所に基づいて決定するために、DISCOには地理空間インデックスが必要となる。
  • geo by demand（需要に基づく地理）。地理空間インデックスは需要にも必要である
  • この情報全てを有効活用するには、より良いルーティングエンジンが必要である。

ディスパッチ

• 車が移動する際、位置更新がgeo by supplyに送信される。乗客をドライバーにマッチングしたり、またはただマップに車を表示したりするのにも、DISCOはgeo by supplyにリクエストを送信する。
• geo by supplyはまず、ざっくりとした通過フィルタを作り、近くにいて、かつ条件に合う候補者を見つける。
• そして、そのリストと条件がルーティング／ETAに送信され、どのくらい近いのかETAが算出される。これは地理的にではなく、ロードシステムによって計算される。
• ETA順にソートされたものがサプライに返され、その情報がドライバーに提供される。
• 空港では、仮想のタクシーの列をエミュレートしなければならない。サプライは、到着する順番を考慮するために、その列に並ばなければならない。

地理空間インデックス

• とても拡張性があるに違いない。設計目標は、毎秒100万の書き込みを処理することだ。書き込み率は、ドライバーから導き出される。ドライバーが移動する際、4秒毎にその情報が送信される。
• 読み込みの目標は、毎秒毎の書き込みより読み込みがさらに多くすることだ。なぜなら、アプリを開いている人は皆、読み込みを行っているからだ。
• 旧地理空間インデックスは、簡略化した仮定を立てることで非常によく機能し、ディスパッチ（派遣）可能なサプライのみをトラッキングしていた。サプライの大部分は何かをしていて忙しく、対応可能な一部のサプライは満たしやすかった。ごくわずかなプロセスで、メモリに保存されたグローバルインデックスがあった。非常に単純なマッチングを行うのは容易だった。
• 新しい環境では、全ての状態のあらゆるサプライがトラッキングされなければならない。それに加え、予測ルートもトラッキングする必要がある。これにより、データはより多くなる。
• 新サービスは何百ものプロセスを実行する。
• 地球は球体だ。純粋に緯度と経度に基づいて要約や概算をすることは難しい。だから、UberはGoogle S2ライブラリを使い地球を小さなセルに分けた。それぞれのセルが一意のセルIDを有している。
• int64を利用すると、地球を平方センチメートルごとに表示することができる。Uberは、地球のどこにいるかによって31㎢から6.38km㎢まである、レベル12のセルを使っている。ボックスは球体のどこにいるかによって、形や大きさを変える。
• S2は形状に必要な範囲を教えてくれる。例えばロンドンを中心に半径1kmの円を描きたかったら、S2はそれを完全にカバーするために必要なセルはどれかを教えてくれる。
• 全てのセルがIDを持っているため、そのIDはシャーディングキーとして使われる。サプライから位置情報が入ってくると、ロケーション用のセルIDが決定される。セルIDをシャードキーとして利用し、サプライのロケーションが更新される。その後、いくつかの複製に送られる。
• DISCOが、あるロケーションに近いサプライを見つける必要がある場合、乗客がいる場所を中心にカバーする円の値が計算される。円面積からのセルIDを使うことで、全ての関連シャードに連絡が渡り、サプライデータが返される。
• これは全て拡張可能だ。あなたの望むように効率的ではないかもしれないが、出力モジュールが比較的安価なので、書き込み性能はノードをさらに増やすことで常に拡張される可能性がある。読み込み性能はレプリカを使うことで拡張される。より多くの読み込み最大容量が必要になれば、レプリケーションファクタが増やされるかもしれない。
• 制約は、セルサイズがレベル12のサイズに固定されていることだ。将来的には、ダイナミックセルサイズがサポートされる可能性がある。トレードオフとして、セルサイズが小さくなるほど、クエリ用の出力モジュールが大きくなる。

ルーティング

• 地理空間の答えが出た後は、選択肢を評価しなければならない。
• これには、ハイレベルな目標がある：
  • 余分な運転を減らすこと。運転は人による仕事なので、より高い利益を上げたいという欲が出てくる。ドライバーは車を余計に乗り回すことで、報酬をもらっている。理想的には、ドライバーは連続して運転に出ることが望ましい。たくさんの仕事が列をなしていれば、その全ての仕事で稼ぐことができる。
  • 待ち時間を減らすこと。乗客の待ち時間は可能な限り少なくすべきだ。
  • 全体的にETA（予定到着時間）を最短にすること。
• 旧システムでは、デマンドが現在受付可能なサプライを検索し、マッチングして対処していた。これは導入しやすく、わかりやすい。そして、個人輸送においては非常によく機能していた。
• 現時点の予約状況を見るだけで、良い選択ができるとは限らない。
• その考えはこうだ。遠くにいる現在アイドリング中のドライバーより、現在乗客を輸送中のドライバーのほうが、今乗車を希望している顧客にとって相応しいかもしれない。運転中のドライバーを選ぶことで、顧客の待ち時間を最小化し、もっと遠くにいるドライバーの運転量を減らすことができる。
• このように先を見通そうとするモデルによって、動的な条件により良く対処することが可能となる。
  • 例えば、顧客の近くにいるドライバーがオンラインになったが、すでにもう一人のドライバーが遠くから派遣されていたとしたら、その決定を変えることはできない。
  • もう一つの例は、顧客が相乗りしても良いという場合だ。非常に複雑なシナリオで未来を予測しようとすることで、さらに最適化することが可能だ。
• このような判断は、宅配便や食料の配達を考慮するともっと面白くなる。その場合、一般的に人は他にすることがあるため、異なる交換条件を伴う。

スケーリング・ディスパッチ

• ディスパッチはnode.jsを使用して構築されている。
• Uberはステートフルサービスを作っているので、スケーリングにステートレスな方法は使えない。
• Nodeは単一プロセスで実行しているため、同じマシンや複数マシンで、Nodeを複数のCPUで実行するメソッドをいくつか考え出さなければならない。
• Erlang全部をJavascriptで再実装するというジョークがある。
• Nodeをスケールさせるための解決策は、ringpopだった。これは、ゴシッププロトコルを用いたコンシステントハッシュリングで、拡張可能でフォールト・トレラントなアプリケーション・レイヤーのシャーディングを実現するものだ。
• CAP定理で言うと、ringpopとは一貫性と引き換えに可用性を手に入れるAPシステムである。機能しないサービスより、不整合を釈明する方がましだろう。たびたびエラーを起こすより、動いている方がいい。
• ringpopは、各Nodeプロセスに含まれている、埋め込み可能なモジュールだ。
• Nodeのインスタンスは、メンバーシップ集合に情報を伝える。一度すべてのノードが互いの存在を認識すれば、単独かつ効率的に検索をし、判断をすることができる。
• これは実にスケーラブルだ。プロセスをもっと追加して、より多くの仕事を片付けよう。シャードデータに使うこともできるし、分散ロッキングシステムとして使うこともできる。また、Pub/Subソケットやlong-pollソケット用の集合ポイントを調整するのにも使えるだろう。
• このゴシッププロトコルはSWIMに基づいている。収束時間改善のために、いくつかの改良がなされた。
• 実行中メンバーのリスト情報が回される。ノードがさらに追加されるので、スケーラブルだ。SWIMの「S」はスケーラブルのSで、実際うまく機能する。今までで数千ものノードを拡張してきた。
• SWIMは、同一プロトコルの一端として、ヘルスチェックとメンバー変更を組み合わせる。
• ringpopシステムでは、これら全てのNodeプロセスにringpopモジュールがある。これらは、現在のメンバーシップに情報を回す。
• 外面的には、DISCOが地理空間を取りたい場合、全ノードは等価である。任意の正常なノードが選択される。リクエストの届いた場所が、ハッシュリング検索を用いて正しいノードにリクエストを転送することを任される。以下のような感じだ。

• これら全てのホップとピアが互いに情報をやり取りしているのは馬鹿げて聞こえるかもしれないが、実に良い特性を生み出してくれる。例えば、どのマシンでも、サービスがインスタンス追加によって拡張される可能性がある。
• ringpopはTChannelと呼ばれるUber独自のRPCメカニズムに基づいて構築されている。
  • これは、TwitterのFinagleにインスパイアされた双方向リクエスト／レスポンスプロトコルだ。
  • 重要な目標は、複数の異なる言語でパフォーマンスを管理することだった。特に、NodeとPythonにおいては、既存のRPCメカニズムの多くはうまく機能しなかった。Redisレベルのパフォーマンスを求めていた。TChannelはすでに、HTTPより20倍速い。
  • 高性能フォワーディングパスが欲しかったので、中間地点は最大のペイロードを把握しなくても非常に簡単に転送決定をすることができた。
  • 適切なパイプラインが欲しかったので、ヘッドオブラインブロッキングがなく、リクエストとレスポンスはいつでも、どの方向にでも送信することができた。全てのクライアントはサーバでもあった。
  • 内蔵型ペイロードチェックサムとトレース機能、ファーストクラス機能が欲しかった。全てのリクエストはシステムを通っていく際、追跡可能であるべきだ。
  • HTTPからのクリーンな移行パスを求めていた。HTTPは、TChannelにごく自然にカプセル化される。
  • UberはHTTPと
    Json事業から撤退しつつある。全てはTChannelとThriftに移りつつある。
• ringpopは持続接続ベースでTChannelにおける全情報のやり取りを行っている。これらの同一持続接続は、アプリケーショントラフィックを広げたり転送したりするために用いられる。TChannelはサービス間の情報伝達にも使用される。

ディスパッチの可用性

• 可用性は非常に重要だ。Uberの競合他社はたくさんあり、スイッチング・コストは非常に低い。もしUberが一時的にでもダウンしたら、その間の売上は他社のところへ行くだろう。他のプロダクトはもっと継続性が有るので、顧客は後でもう一度試してみるかもしれない。Uberの場合、そうとも限らない。
• 全て再試行可能にする。うまく機能しないものは、再試行可能でなければならない。これが不具合を迂回する秘訣だ。これには、全てのリクエストが冪等である必要がある。例えば派遣を再試行する場合、ドライバーを2回派遣したり、誰かのクレジットカードに2度請求したりすることがあってはならない。
• 全てを強制終了可能にする。不具合はよくあるケースだ。処理の強制終了でやみくもに損害を与えるべきではない。
• クラッシュ時に限る。システムをシャットダウンするのに正しい手順に従うことなどない。正常終了が実践される必要性はない。その必要があるのは予想外の出来事が発生した時だ。
• 細かくする。起こっている不具合のコストを最小化するには、その物事を細分化することだ。１つのインスタンスでグローバルトラフィックを処理することは可能かもしれないが、それが機能しなくなった時どうなるのだろうか？インスタンスが1組ある場合、一つが不具合を起こしてもキャパシティは半分になる。だから、サービスは分割されている必要があるのだ。これは、技術的な問題のように聞こえるが、むしろ文化的な問題だ。データベースを1組用意するほうがより簡単だ。用意するのは当たり前のことでも、1組となると悪になる。1つを自動プロモートし、新しく代理データベースを再起動することが可能なのであれば、やみくもに2つのデータベースを強制終了させるのは非常に危険だ。
• 全て強制終了する。そういった不具合を確実に切り抜けることができるよう、データベースを全て強制終了するくらいのことはしよう。そうするには、どのデータベースを使うかを適宜判断する必要がある。同社はMySQLの代わりにRiakを選んだ。同様に、redisの代わりにringpopを使っている。redisのインスタンスを強制する作業は高くつく。通常非常に大規模ので、消すにはたくさんの費用がかかる。
• 小さな塊に分ける。文化面での改革の話だ。一般的に、ロードバランサを介してサービスAがサービスBと通信をする。ロードバランサが機能しなくなったらどうなるだろうか？あなたならそれにどう対処するだろうか？その方法を使ってみたことがなければ、分からないだろう。だから、ロードバランサを強制終了してみるべきだ。あなたはどうやってロードバランサを避けて通るだろうか？ロードバランシングのロジックは、サービスそれ自体の中に置かれるべきだ。クライアントには、問題を避ける方法を知る知能が求められる。これは、哲学的にFinagleの仕組みと似ている。
• システム全体をスケールし、バックプレッシャーに対処するため、ringpopeノード群からサービスの発見とルーティングシステムが生み出された。

総合データセンターの不具合

• これは頻繁に起こることではないが、不具合が立て続けに起こったり、上流のネットワークプロバイダが不具合を起こしたりすることがあるかもしれない。
• Uberはバックアップデータセンターを設置し、全てバックアップデータセンターに転送する切り替え環境が整っている。
• 問題は、進行中の輸送用データがバックアップデータセンターにないかもしれないということだ。Uberでは、データを複製するのではなく、むしろ運転手の電話を輸送データのソースとして利用している。
• 現状では、ディスパッチシステムが、暗号化されたステートダイジェスト（状態の要約）を定期的にドライバーの電話に送信している。さて、仮にデータセンターのフェイルオーバーがあったとしよう。次に、ドライバーの電話がディスパッチシステムに位置更新を送信する。するとディスパッチシステムは、今回の輸送についての情報がないと判断してステートダイジェストに問い合わせる。その後、ディスパッチシステムはステートダイジェストからの情報を更新し、何事もなかったかのように輸送が続けられる。

否定的側面

• Uberの拡張性と可用性の問題に関する解決策のマイナス面は、Nodeプロセスが互いにリクエストを転送し合うことと、複数の出力モジュールがメッセージを送信することで、レイテンシが高くなる可能性があることだ。
• 出力モジュールシステムでは、些細な誤作動や異常が驚くほど大きな影響をもたらす。一つのシステム内の出力モジュール数が多いほど、高レイテンシのリクエストを得る可能性が高くなる。
• 良い解決策は、クロスサーバでのキャンセレーションを行い、バックアップ要求をすることだ。これは、TChannelにファーストクラス機能として内蔵されている。サービスB（2）にもリクエストが送られるという情報とともに、サービスB（1）にリクエストが送信される。少し遅れて、リクエストがサービスB（2）に送られる。B（1）がリクエストを完了したら、B（2）へのリクエストをキャンセルする。多くの場合、遅れによりB（2）は何もしていない状態だ。しかし、B（1）が失敗した場合、B（2）がリクエスト処理をし、返答する。この時のレイテンシは、B（1）が最初に試みて、タイムアウトが起こってからB（2）が試した場合よりも低い。
• これについての詳しい情報は、Google On Latency Tolerant Systems: Making A Predictable Whole Out Of Unpredictable Partsという記事を参照してほしい。

原文：http://highscalability.com/blog/2015/9/14/how-uber-scales-their-real-time-market-platform.html(2016-4-27)
※元記事の筆者には直接翻訳の許可を頂いて、翻訳・公開しております。

 -Techエンジニア, テックブログ