Facebook、Hadoopのスケーラビリティ問題を解決する、独自の「Hadoop Corona」をオープンソースで公開

Facebookでは、24時間ごとに0.5ペタバイトのデータが生成され、それらを分析するために毎日6万回以上ものHiveのクエリが実行されているそうです。

こうした大規模処理を行うプラットフォームとして使われているのがHadoop。しかしFacebookはHadoop MapReduceのスケーラビリティに限界を感じており、それを解決するための新しいソフトウェア「Hadoop Corona」を開発、オープンソースで公開しました。

Facebookのページ「Under the Hood: Scheduling MapReduce jobs more efficiently with Corona」では、従来のHadoop MapReduceのどこに課題があったのか、4つに分けて指摘しています。まずはそれを見ていきましょう。

ジョブトラッカーの設計上の問題によってスケーラビリティの制限がある、というのが1つ目の指摘です。

Specifically, we began to see issues with the scheduling framework, which consists of a job tracker and many task trackers (one for each worker machine). Task trackers are responsible for running the tasks that the job tracker assigns them.

とりわけ、スケジューリングのフレームワークの課題が目につき始めました。これはジョブトラッカーと多くのタスクトラッカーから構成されています。タスクトラッカーはタスクの実行を行い、ジョブトラッカーはそららのタスクをアサインします。

The job tracker has two primary responsibilities: 1) managing the cluster resources and 2) scheduling all user jobs. As the cluster size and the number of jobs at Facebook grew, the scalability limitations of this design became clear. The job tracker could not handle its dual responsibilities adequately. At peak load, cluster utilization would drop precipitously due to scheduling overhead.

ジョブトラッカーは2つの主要な仕事を担っています。（1）クラスタリソースの管理、（2）すべてのユーザージョブのスケジューリング。Facebookのクラスタサイズとジョブが増加するにつれ、この設計によるスケーラビリティの制限がはっきりしてきました。

ジョブトラッカーはこの両方の仕事を適切に扱えないのです。負荷がピークに達した時点で、クラスタの利用効率はスケジューリングのオーバーヘッドによって突如下落します。

2つ目の指摘は、スケジューリングの問題。

Another limitation of the Hadoop MapReduce framework was its pull-based scheduling model. Task trackers provide a heartbeat status to the job tracker in order to get tasks to run. Since the heartbeat is periodic, there is always a pre-defined delay when scheduling tasks for any job. For small jobs this delay was problematic.

もう1つのHadoop MapReduceフレームワークの制限は、プルベースのスケジューリングモデルです。タスクトラッカーはジョブトラッカーにハートビートを提供し、これによって実行すべきタスクが取得されます。ハートビートは定期的に発生するものであり、そこにはあらゆるジョブのタスクスケジュール時に一定程度の事前遅延が生じます。小さなジョブにおいて、この遅延は問題となります。

3つ目の指摘は、リソース管理の問題。

Hadoop MapReduce is also constrained by its static slot-based resource management model. Rather than using a true resource management system, a MapReduce cluster is divided into a fixed number of map and reduce slots based on a static configuration – so slots are wasted anytime the cluster workload does not fit the static configuration. Furthermore, the slot-based model makes it hard for non-MapReduce applications to be scheduled appropriately.

Hadoop MapReduceは静的なスロットベースのリソースマネジメントモデルによっても制約されています。これはむしろ、リソース管理システムというよりも、静的なコンフィグレーションによって設定されたMapとReduceのスロットに分割されることによります。つまりワークロードが静的なコンフィグレーションに合致しない限り、スロットはつねに無駄が生じているのです。

そして4つ目の指摘は、ダウンタイムの問題です。

Finally, the original job tracker design required hard downtime (all running jobs are killed) during a software upgrade, which meant that every software upgrade resulted in significant wasted computation.

最後に、オリジナルのジョブトラッカーの設計では、ソフトウェアアップグレード時に厳密なダウンタイム（すべての実行中のジョブがKillされた状態）を必要とします。アップグレードのときにはいつも、多大なコンピュータリソースを遊ばせてしまうことになるわけです。

Hadoop Coronaでスケーラビリティと利用効率を改善

Facebookでは、こうした課題を解決するものとして、独自にHadoop Coronaを開発。これによって主に以下の4つの改善が行われたと説明しています。

• より優れたスケーラビリティとクラスタの利用効率
• 小さなジョブにおける低レイテンシ
• 厳密なダウンタイムを必要としないソフトウェアアップグレード
• 実際のタスクのリソース要求を元にしたスケジューリング

設計上の特徴は、クラスタのリソースマネジメントをジョブトラッカーから切り離したこと。

To address these issues, we developed Corona, a new scheduling framework that separates cluster resource management from job coordination.[1] Corona introduces a cluster manager whose only purpose is to track the nodes in the cluster and the amount of free resources. A dedicated job tracker is created for each job, and can run either in the same process as the client (for small jobs) or as a separate process in the cluster (for large jobs).

こうした問題を解決するために、私たちはCoronaを開発しました。新しいスケジューリングフレームワークを備え、クラスタのリソースマネジメントをジョブコーディネーションから切り離しました。Coronaはクラスタマネージャを導入。この唯一の目的は、クラスタのノードとフリーリソースの量を追跡することです。ジョブごとに専用のジョブトラッカーが生成され、同じプロセス内で（小さなジョブ用に）クライアントとしても、（大きなジョブ用に）クラスタ内の別プロセスとしても、どちらでも走らせることができます。

スケジューリングをプル型からプッシュ型に変えたことで、レイテンシの改善も図ったと説明されています。

One major difference from our previous Hadoop MapReduce implementation is that Corona uses push-based, rather than pull-based, scheduling. After the cluster manager receives resource requests from the job tracker, it pushes the resource grants back to the job tracker. Also, once the job tracker gets resource grants, it creates tasks and then pushes these tasks to the task trackers for running. There is no periodic heartbeat involved in this scheduling, so the scheduling latency is minimized.

従来のHadoop MapReduceの実装との大きな違いのひとつは、Coronaはプルではなくプッシュベースのスケジューリングを採用していることです。クラスタマネージャがジョブトラッカーからリソース要求を受け取ると、リソースの許可をジョブトラッカーに返します。ジョブトラッカーが許可を得ると、タスクを生成し、実行のためタスクトラッカーへタスクをプッシュします。ここにはスケジューリングに関わる定期的なハートビートはなく、スケジュールのレイテンシは最小で済みます。

Facebookでは、まず500ノード程度の試験導入からはじめ、続いてクリティカルでないワークロードを移行、その次にすべてのMapReduceのジョブをCoronaへ移し替える、という作業をすでに今年の半ばまでに終えているとのこと。

CoronaはGithubですでに公開されています。

さまざまな大規模分散処理プラットフォームが共存

Hadoopの進化形としては、すでに元MapReduce 2.0、いまYARNという名前になったプロジェクトがあります。このYARNもMapReduceのジョブトラッカーとタスクトラッカーの構成を変え、スケーラビリティの向上を実現しようとしていますが、Coronaとは別物です（YARNについては「日々進化するHadoop。これまでのおさらいと最近の動向（後編）」で紹介しています）。

またHadoopのディストリビューションベンダとして知られるClouderaは、MapReduceよりも何倍も高速でSQLにも対応した分散クエリエンジン「Cloudera Impala」を発表しています。Apacheでも同様のApache Drillの開発が進められています。

Hadoopの登場で急速に注目度が高まった大規模分散処理プラットフォームは、さまざまなソフトウェアが登場し、共存することになっていきそうです。

（ちなみにYARNについてFacebookでは「we found numerous incompatibilities that would be time-prohibitive and risky to fix. Also, it is unknown when YARN would be ready to work at Facebook-scale workloads.」（互換性がないし、いつ利用可能になるかも分からない）と書いています）