YT：Yandexが独自のMapReduceシステムを必要とする理由とその仕組み

過去6年間、Yandexはコード名YT（ロシア語では「Yt」と呼ばれる）のシステムに取り組んできました。 これは、大量のデータを保存および処理するための主要なプラットフォームです。YaC2013で既に説明しました。それ以来、開発を続けています。 今日は、YTの開発がどのように始まったのか、YTにどのような新しいものが登場したのか、また近い将来に私たちがやろうとしていることについてお話します。







ちなみに、10月15日にYandexオフィスで、YTだけでなく、他のインフラストラクチャテクノロジーであるMedia Storage、Yandex Query Language、 ClickHouseについても話します。 会議では、秘密を明らかにします-YandexにあるMapReduceシステムの数を教えます。

どのような問題を解決しますか？

活動の性質上、Yandexは、通常のユーザーが処理する必要のないボリュームのデータを保存および処理する必要性に常に直面しています。 検索ログとインデックス、ユーザーデータ、地図作成情報、中間データ、機械学習アルゴリズムの結果-これらはすべて、数百ペタバイトのディスクスペースを占有します。 このようなボリュームを効率的に処理するために、MapReduceパラダイムが伝統的に使用されています。これにより、コンピューティング効率とユーザーコードのシンプルさのバランスが取れています。



何らかの形で、MapReduceに似た計算は、Google、Microsoft、Facebook、Yahooなどの大手ビッグデータ企業すべてで使用されています。 さらに、はるかに控えめなサイズのデータ​​を操作する場合、事実上のMapReduceは長い間標準になりました。特に、Hadoop MapReduceフレームワークでのオープンソース実装の成功と、MapReduceの上に便利な分析ツールを提供する開発されたエコシステムの存在のためです。



もちろん、すべてのタスクがMapReduceモデルに成功するわけではありません。 主な制限は、大量のデータが変換のチェーン全体を通過するバッチ実行を伴うことです。 これらの変換には数分、数時間、時には数日かかります。 この作業方法はユーザーへのインタラクティブな応答には適していないため、MapReduceの計算に加えて、従来はKVストレージ（キーと値のストレージ）などのシステムを使用します。 このようなシステムは世界中に数多くあります：数十テラバイトのデータを保存するのに適したクラシックRDBMS（Oracle、Microsoft SQL Server、MySQL、PostgreSQL）から始まり、多くのペタバイトで動作するNoSQLシステム（HBase、Cassandra）で終わります。



MapReduceの場合、同時に使用されるいくつかの異なるシステムの存在はあまり意味がありません。 事実、それらすべてが実際に同じ問題を解決するということです。 ただし、KVストレージの場合、解決されるタスクの多様性の程度と一般的なレベルの特異性ははるかに高くなります。 Yandexでは、このようなシステムを一度にいくつか使用します。 データの一部はMySQLに保存されます。MySQLは、信頼性とさまざまな機能の面で実証されています。 データ量が多く、処理速度が重要な場合、Yandex独自のDBMSであるClickHouseを使用します 。 最後に、過去3年間にわたり、YTのフレームワーク内でKVストレージを作成するタスクに積極的に取り組んできました。 高い信頼性のストレージ、トランザクションサポート、MapReduce計算との統合が必要な場合に使用します。

Hadoopスタックを使用しないのはなぜですか？

ビッグデータインフラストラクチャの大部分には、同様の問題を解決する公開製品があります。 もちろん、まず、Hadoopエコシステムについて話します。これに基づいて、理論的には、分散ファイルシステム（HDFS）およびMapReduceフレームワークからKVリポジトリ（HBase、Cassandra）までの計算スタック全体を構築できます。分析（Hive、Impala）およびデータ処理システム（Spark）。



ビッグデータを扱う中小企業にとって、Hadoopは多くの場合、競合しない選択肢です。 大規模なプレーヤーは、Hadoopを積極的に使用および開発するか（たとえば、YahooやFacebookがこれを行いました）、独自のインフラストラクチャ（Google、Microsoft）を作成します。



これら2つの可能性を選択して、Yandexは2番目の方法を採用しました。 この決定は容易なことではなく、もちろん、唯一の正しい決定とみなすことはできません。 多くの要因が考慮されました。 最も重要なものを簡単に要約してみましょう。

• 歴史的に、Yandexでは最大量の計算コードがC ++で記述されています。 同時に、Hadoopは基本的にJavaプラットフォームを指向しています。 これ自体は致命的ではありません-C ++の古いコードをサポートし、Javaで新しいコードを開発するためのアダプターをいつでも作成できます。 ただし、長年の開発を通じて、C ++での作業において同社がかなりの経験を積んできたこと、そしてJava開発がより控えめに提示されたことを認識する価値があります。
  
  
• YahooやFacebookなどの企業でのHadoopの使用成功例は、Hadoopを効果的に使用するために通常はバニラバージョンは使用されないことを示していますが、フォークはその後の高度な改良とともに実行されます。 作成するには、Hadoopのコア開発者を社内に配置することが非常に望ましい、または必要ですらあります。 これらの専門家は、アメリカのIT市場の基準でも非常にまれなので、Yandexにとって、十分な数の人材を雇うことは非常に難しい作業です。
  
  
  
  
• おそらく、プラットフォームの上記のフォークはメインの開発ブランチに決して注ぎ込まれず、このブランチと互換性がなくなるでしょう。 既製のエコシステムソリューションを使用する利点は、2つのツールセットに限定されます。これらのツールは、分離時に「持ち歩く」ことができ、フォークを介して転送できます。
  
  
  
  
• しかし、なぜHadoopをフォークするのが面倒なのでしょうか？ 可能な場合は、開発に参加して、公開されており進化しているバージョンを使用しないのはなぜですか？ 実際のところ、Hadoop自体はかなり優れていますが、完璧にはほど遠いということです。 Yandexの自社インフラストラクチャでの作業中に、システムに非常に根本的な変更を加えなければ解決できない問題に繰り返し直面しました。 Hadoopの開発には十分な能力がないため、公的な支店にこのような大幅な変更を加えることはできません。 このような過激なパッチは、ほぼ確実に妥当な時間内にコードレビューに合格しません。
  
  
  
  
• また、Hadoopを社内に完全に統合するには、このプラットフォームに高度な資格を持つスペシャリストを配置する必要はありません。たとえば、Clouderaなど、Hadoopソリューションの大規模なインテグレーターのサービスに限定すれば十分です。 ただし、活動の種類ごとに私たちに近い企業の常識と経験は、このようなヘルプは実装の最初の段階では役立つことがありますが、Hadoopコードベースの大幅な変更を必要とする複雑なタスクは、個別に解決する必要があることを示しています（たとえば、 メールエクスペリエンスを参照してください） 。 RU ）。

もちろん、既製のHadoopツールを再利用することを拒否すると、Hadoopコミュニティと競合する方法がなく、世界中の何百、何千もの開発者を数えることができないため、作業が非常に複雑になりました。 一方、上で示したように、Yandexのエコシステムにとって重要な変更を合理的な価格で得ることができないことと、Hadoopを使用している企業の経験は、すべての利点を上回りました。



私たちが行った選択に基づいて、インフラストラクチャの最も関連性の高い部分に積極的に取り組み、YTプラットフォーム自体とその上に作成されたツールといった少数の主要なシステムの改善に主な努力を注いでいます。

私たちは今何を知っていますか？

上記から、YTは少なくともMapReduceシステムであることがわかります。 本当ですが、すべてではありません。 YTは誕生以来、ビッグデータの分野における幅広い計算上の問題を解決するための基礎となることを目標に設計されました。



YaCに関するYTレポートから3年で、システムは大幅に開発されました。 その後、言及された問題の多くは正常に解決されました。 ただし、会社（およびインフラストラクチャ）が積極的に開発されているため、退屈していません。 そのため、新しいタスクが毎日届きます。



YTの現在のアーキテクチャを簡単に説明しましょう。

クラスター

YTは、デプロイメントユニットがクラスターであるマルチコンポーネントシステムです。 現在、最大10,000台のサーバー（ ノード ）の合計数を持つクラスターをサポートしています。この台数のマシンには、数百ペタバイトのデータを保存できるだけでなく、数十万のCPUコアで計算を計画および実行できます。



YTクラスターのほとんどは、完全に同じデータセンター内にあります。 これは、MapReduceの計算プロセスでは、マシン間で同種のネットワークを持つことが重要であるためです。これは、地理的に離れているとスケールで達成することはほとんど不可能です。 このルールには例外があります。内部データフローが小さい一部のYTクラスターは、それにもかかわらず複数のDC間で分散されます。 したがって、それらの1つまたは複数の完全な障害は、システムの動作に影響しません。

DFS

YTの最下層は、分散ファイルシステム（DFS）です。 初期のDFS実装は、従来のHDFSまたはGFSに似ていました。 データは、 チャンクと呼ばれる小さな（通常は最大1 GB）フラグメントの形でクラスターノードディスクに保存されます。 システムで使用可能なすべてのチャンクに関する情報、DFSに存在するファイルとテーブル、およびそれらがどのチャンクで構成されるかは、 マスターサーバーのメモリにあります 。



DFS の名前空間 （ディレクトリ構造、ファイル名とテーブル名、メタ属性、アクセス許可など）は、 Cypressと呼ばれます。 ほとんどの従来のファイルシステムとは異なり、サイプレスはトランザクション性をサポートしています。つまり、構造とデータを使用して複雑な操作をアトミックに実行できます。 たとえば、サイプレストランザクションのフレームワーク内では、いくつかのファイルをシステムにアトミックにアップロードしたり、データの新しい部分がメタ属性に到着したことを記録したりすることができます。 サイプレスは、ノードのロックシステムもサポートしています。 YTクラスターの1つはデータセンター間で分散され、実際にはチャンクを格納しませんが、そのサイプレスは非常にアクセスしやすいメタデータストレージおよび調整サービスとしてさまざまなインフラストラクチャプロセスで使用されます（Chubby、Zookeeperまたはetcdと同様）。







クラスタを構成するマシンの数を考えると、機器の故障は毎日発生し、固有の現象ではありません。 システムは、人間の介入なしで、自動モードでほとんどのタイプの典型的な障害に耐えるように設計されています。 特に、システムにはいわゆる単一障害点 （SPoF、単一障害点）がありません。つまり、1台のサーバーを壊してもシステムの動作を中断することはできません。 さらに、実用的な観点からは、より大きなランダムサブセット（たとえば、クラスター内のランダムに選択された2つのサーバー）またはマシンの構造グループ全体（ラックなど）で障害が発生した場合でも、システムが動作し続けることが重要です。



分散DFSの場合、マスターサーバー（存在する場合）は自然なSPoFを形成します。これは、電源がオフになるとシステムがデータの送受信を停止するためです。 そのため、最初からマスターサーバーが複製されていました 。 最も単純なケースでは、1つのマスターサーバーの代わりに、同じ状態が維持される3つのマスターサーバーがあると想定できます。 3つのマスターサーバーのいずれかに障害が発生しても、システムは動作を継続できます。 このような切り替えは完全に自動的に行われ、ダウンタイムは発生しません。 マスターサーバーのセット間の同期は、 コンセンサスアルゴリズムの種類の1つを使用して提供されます。 この場合、独自のデザインのライブラリを使用します。これはHydraと呼ばれます。 Hydraの最も近いオープンソースアナログは、Raftアルゴリズムです。 Apache Zookeeperシステムでは、綿密なアイデア（ZABアルゴリズム）が使用されます。







3つの複製されたマスターサーバーを使用した説明したスキームは、フォールトトレランスの観点からは良好ですが、必要なボリュームにクラスターをスケーリングすることはできません。 実際、メタデータの量に関しては、別のマスターサーバーのメモリによって制限されたままです。 そのため、約1年前に、マルチセルと呼ばれるユーザー透過的なチャンクメタデータシャーディングシステムを実装しました。 それが使用されると、メタデータはマスターサーバーの異なるトリプレット間で共有されます（彼らが言うようにshard ）。 各トリプレット内で、Hydraは引き続き機能し、フォールトトレランスを提供します。



HDFSでのチャンクメタデータのシャーディングは現在提供されていないことに注意してください。 代わりに、利便性の低いHDFSフェデレーション （疑わしいユーティリティのプロジェクト）を使用することを提案します。これは、少量のデータでは意味がなく、レポートによると、大企業での使用経験は必ずしもプラスではないためです。



もちろん、フォールトトレランスは、メタデータのレベルだけでなく、チャンクに格納されているデータのレベルでも重要です。 これを実現するには、2つの方法を使用します。 1つ目はレプリケーションです 。つまり、異なるサーバー上の複数（通常は3つ）のコピーにある各チャンクのストレージです。 2番目は消去コーディングです 。つまり、各チャンクを少数の部分に分割し、チェックサムを使用して追加の部分を計算し、これらすべての部分を異なるサーバーに保存します。







後者の方法は特に興味深いです。 ボリュームXの保存に通常のトリプルレプリケーションを使用する場合、追加のレプリカにさらに2Xを費やす必要があり、適切な消去コードを使用すると、このオーバーヘッドはX / 2に削減されます。 私たちが対処しなければならない量（数十ペタバイト）を考えると、その増加はとてつもなく大きいです。 既存のシステムのフレームワーク内で、消去コーディングは約4年間利用可能です。 Hadoopでの同様のサポート（いわゆるEC-HDFS）が利用可能になったのは1年前に過ぎないことに注意してください。



テーブル







ユーザーに表示されるメインストレージユニットがファイルであるHadoop MapreduceのHDFSバンドルとは異なり、YT DFSファイルは存在しますが、アクティブに使用されることはありません。 代わりに、データは主にテーブル （データの列で構成される行のセット）に保存されます。 システムの開発中、テーブル内の効率的でコンパクトなデータストレージを実現するために多大な労力を費やしました。 ストレージフォーマットフォーマット（Hadoop ORC、Parquet）のアイデアと、さまざまなブロック圧縮アルゴリズム（lz4、gzip、brotliなど）を組み合わせています。



ユーザーにとって、このような実装の詳細は透過的です。彼は単にテーブルレイアウト（列の名前とタイプを示す）、目的の圧縮アルゴリズム、複製方法などを設定します。Hadoopユーザーは、特定のフォーマットのデータを含む、YTユーザーは、テーブル全体とその部分の用語を使用します。並べ替え、マップ、縮小、結合操作を実行し、行範囲ごとにテーブルを読み取ります。

MapReduceスケジューラ

格納するデータのほとんどの計算は、MapReduceモデルで実行されます。 既に述べたように、クラスターの計算能力の合計プールは数十万コアです。 それらの使用は中央プランナーによって制御されます。 map、sort、reduceなどの大きな計算（ 操作と呼ばれる）を個別のローカルパーツ（ ジョブ ）に分割し、それらの間でリソースを分配し、実行を監視し、障害が発生した場合に計算の一部を再開することでフォールトトレランスを保証します。







計画のための2つの主要なリソースであるCPUとメモリを区別します。 オペレーション間でリソースを分離するために、非常に柔軟なスキームである階層ドミナントリソースフェアシェアを使用します。これにより、計算をツリー構造にグループ化し、別々のサブツリーに制限（下限と上限）を設定できます。 特に、単一のクラスターのフレームワーク内で、さまざまなプロジェクトが生産回路とテスト回路の両方を同時に保持します。 同じクラスターで、一部のリソースがアドホック分析に割り当てられます。



一般的な大規模なYTクラスターでは、1日で25万件を超える操作が実行され、合計で1億のジョブが実行されます。



現時点では、クラスター内のアクティブなスケジューラーは常に1つですが、フォールトトレランスのために、システムには常にパッシブスタンバイレプリカがあります。 メインスケジューラが10秒間のオーダーで失敗すると、その別のレプリカに切り替わります。 後者は、計算操作の状態を復元し、実行を継続します。



現在、スケジューラは、クラスタ全体で最もクリティカルにロードされた単一のマシンです。 スケジューラコードは、同時に管理する必要があるジョブ（数十万）と操作（数千）の数を考慮して、繰り返し最適化されています。 高度な並列処理を達成できたため、実際には1台のマシンのリソースを完全に使用しています。 ただし、クラスターサイズのさらなる成長のために、このアプローチは行き止まりのように思われるため、新しい分散スケジューラの作成に積極的に取り組んでいます。



現在、YT MapReduceレイヤーは、何百ものYandex開発者、アナリスト、マネージャーが毎日使用して、データの分析計算を実行しています。 また、多数のさまざまなプロジェクトがそれを使用し、メインサービスのデータを自動的に準備します。

KVストレージ

既に何度か述べたように、MapReduceの計算は、私たちの前のタスクの全範囲をカバーしていません。 従来のMapReduceテーブル（ 静的と呼ばれます ）は、大量のデータを効率的に保存し、MapReduce操作で処理するのに最適です。 一方、KVストレージを置き換える方法ではありません。 実際、静的テーブルからの行の一般的な読み取り時間は数百ミリ秒単位で測定されます。 そのようなテーブルへの書き込みはそれだけでなく、最後にのみ可能であるため、対話型サービスを作成するためにそのようなテーブルを使用することについて話す必要はありません。



約3年前、YTの一部として、HBaseやCassandraなどのシステムで見られるものと同様の新しい動的テーブルの作成に取り組み始めました。 これを行うには、ベースレイヤー（Hydra、DFS）に多くの改善を行う必要がありました。 その結果、共通の名前空間MapReduceのフレームワーク内でユーザーの動的テーブルを作成する機会が本当にあります。 これらは、文字列および主キーの範囲に対する効率的な読み取りおよび書き込み操作をサポートします。 私たちのシステムは、 トランザクション性と厳密な一貫性 （いわゆるスナップショット分離レベル）、および分散トランザクションのサポート、つまり、1つのトランザクションで1つのテーブルの複数の行または複数の異なるテーブルをアトミックに変更する機能によって区別されます。







最も近い類似物の観点から、私たちの動的テーブルはBigTableとHBaseの古典的なアイデアに近いですが、より強力なトランザクションモデルをサポートするように適合されています。 同時に、同様の基本原則にもかかわらず、パッチをコミットして既存のオープンソースシステム（HBaseなど）に同様の機能を追加することは現実的ではありません。 （たとえば、Facebookで開発中のHydraBaseプロジェクトのコードの一部をHBaseに移植することを説明したチケットを参照してください。例は、地域サーバーのホットレプリケーションによりシステムの可用性を高めるHydraBaseの機能がアプリケーションにとって非常に重要であることを特に示しています。 YT内で彼女をサポートしています。）



これらのテーブルの上に、フィルタリング、集計、最も単純な種類の結合などのインタラクティブなクエリを分散的に実行する機能もあります。 （公的に利用可能なシステムの中で、そのような機会は、たとえばImpalaやPrestoによって提供されます。）Yandexでは、動的テーブルの最大のユーザーはバナーシステムです。 適切な分散KVリポジトリと同様に、YTは1秒あたり数百万の読み取りおよび書き込み操作を処理できます。

私たちの計画は何ですか？

上記の説明からでも、システムが積極的に開発されていることは容易に理解できます。 確かに、私たちが直面しているタスクの規模と野心を考えると、明確で十分に測定可能な利益を受けて、改善できなかったシステムの少なくとも1つの側面を考え出すことは困難です。



最後の部分では、YTの一部として取り組んでいる多くの長期開発プロジェクトに注目する価値があります。



データセンター間レプリケーション



Yandex規模の企業は、単一のデータセンターに依存する余裕はありません。実際、YTクラスターは一度にいくつか展開されます。 それらの間でデータを同期するタスクは複雑で重要です。なぜなら、フォールトトレランスに関する懸念の一部を高レベルのサービスからメインインフラストラクチャに移行できるからです。



格納されたデータボリュームが小さい場合、従来、データセンター全体の損失に耐えることができるYTクロスデータセンターインストールを展開します。 一方、このアプローチには多くの欠点があります。 特に、このようなデータセンター間インストールの開発と更新は非常に重要なタスクです。 実際には、可能な限り100％のアップタイムを提供する必要があります。



各データセンターに1つずつある、半独立クラスターのセットからのシステムは、運用の観点からはより便利に見えます。 そのため、多くの場合、静的プロセス間でデータを複製するための自動プロセスが構成された個別のクラスターを使用します。



また、動的テーブルを複製することも学習します。これにより、さまざまな複製モード（同期、非同期）および定足数記録を一連のデータセンターに実装できます。



マスターサーバーのさらなるスケーリング



そのため、YTでは、チャンクに関連するマスターサーバーのメタデータの一部を分割する方法を学びました。 しかし、ほとんどの場合、ユーザーはチャンクではなく、ファイル、テーブル、ディレクトリのツリー、つまりサイプレスで直接作業します。 このツリーはまだ1つのマスターサーバー（レプリケーションを考慮に入れて3つ）のメモリに格納されており、当然そのノードの数は約1億のレベルに制限されます。ユーザー数が増加し、クラスターが拡大するにつれて、おそらくシャーディングの問題を解決する必要がありますこのメタデータ。



静的テーブルと動的テーブルの統合



最初に動的テーブルがMapReduceモデルと静的テーブルから「離れて」作成されたという事実にもかかわらず、これらのツールを1つの計算で組み合わせることができるという利点がすぐに明らかになりました。



たとえば、MapReduceの長いチェーンを構築し、サービスのインデックスを使用して静的テーブルを計算し、その後、少し手を動かすだけで動的テーブルに変換し、メモリにロードして、ミリ秒単位で測定される待機時間でサービスのフロントエンドに対象データを収集する機能を提供できます。



または、複雑なMapReduce計算の実行中に、いわゆるマップ側結合を実行することにより、参照データを動的テーブルに適用することが可能だとしましょう。



YTのフレームワーク内でのダイナミクスとスタティックの統合は、大規模で複雑なプロジェクトであり、まだ完成にはほど遠いですが、従来は困難と考えられていた多くの問題の解決を根本的に簡素化および最適化できます。



クラウドパワー



Yandexサーバー群の大部分が、検索インデックスデータを直接保存し、ユーザーのリクエストに応答するマシンで構成される、いわゆる検索クラスターで構成されることは周知の事実です。 このようなマシンの負荷プロファイルは非常に特徴的です。 メインのMapTeduceクラスターYTの場合、CPUがほとんど常にボトルネック（ほぼ100％の負荷を24時間年中無休で表示）である場合、検索クラスターのCPU負荷は、たとえば時刻に応じて著しく変化します。



これにより、解放された電力を使用してMapReduce計算を実行できる可能性が広がります。 これにより、サイズが動的に変化する検索クラウドに部分的に転送されます。 このような追加容量の効果的な使用は、私たちにとってもう1つの優先プロジェクトです。 その成功のためには、クラスター管理システム、コンテナ化、プロジェクト間で共有されるリソースの共有（メモリ、CPU、ネットワーク、ディスクなど）を開発する必要があります。