分散システムを学ぶにはどうすればいいですか? 1つの記事ですべてを入手できます!

インターネット企業における分散システムの適用は非常に一般的になり、オープンソース ソフトウェアが次々と登場しています。 Hadoop エコシステム、HDFS から HBase、MapReduce から Spark、Storm から Spark Streaming、Heron、Flink など、オープンソースの海で迷子にならないようにするにはどうすればよいでしょうか?この記事では、基本的な概念とアーキテクチャから分散システムを学ぶ方法を、私自身の学習と仕事の経験と組み合わせて説明します。分散システムの理論体系は非常に大きく、知識も広範囲にわたるため、著者の能力には限界があります。何か不足な点がありましたら、お気軽に話し合い、意見を交換してください。

一般的な分散システムは、HDFS や HBase などのデータ ストレージ システムに分けられます。 Storm、Spark、Flink などのデータ処理およびコンピューティング システム。 Elastic Search や Druid など、データ ストレージに基づく複雑なデータ検索およびクエリ機能を提供する、データ ストレージと分析のハイブリッド システム。ストレージ システムとコンピューティング システムについては、個別に分析することもできます。そのため、この記事では、データ ストレージとコンピューティングの 2 つのシステムについて説明します。

この記事の全体的な構成は次のとおりです。パート I、分散システムの基本概念。パート II とパート III では、それぞれデータ ストレージ システムとデータ コンピューティング システムについて詳しく説明します。最後の部分は要約です。

コンセプト

分散システム:誰もが分散システムについて話していますが、分散システムとは何でしょうか?基本的な概念は、コンポーネントがネットワーク コンピューター全体に分散され、コンポーネントはメッセージの受け渡しを通じてのみ通信し、アクションを調整するというものです。

分散システムとは、ネットワーク化されたコンピュータに配置されたコンポーネントが、メッセージを渡すことによってのみ通信し、動作を調整するシステムです。 (分散システムの概念と設計からの抜粋)

• ノード: ノードは、上記の概念で説明したコンポーネントとして理解できます。実際、これはサーバー上の独立したプロセスに対応する、完全なロジックのセットを完了するプログラム エンティティです。ノードに関しては、ノードがステートフルかステートレスかを検討します。判断基準は単純です。独立したノードはローカルに保存されたステータス情報の一部を維持しているか、またはノードの動作を以前のものと一貫性を保ちながらいつでも他のサーバーに移行できるかということです。そうであれば、ノードはステートレスであり、そうでない場合はステートフルです。
• 例外: 例外処理は分散システムの中核的な問題と言えます。では、分散例外処理とスタンドアロン例外処理の違いは何でしょうか?スタンドアロン システムでは、プログラムの処理結果は予測可能であり、成功か失敗かを問わず、結果は非常に明確です。ただし、分散環境では、処理結果には成功または失敗が明確に返されるだけでなく、タイムアウトという別のステータスもあります。タイムアウトは、処理結果が完全に不確実であることを意味します。正常に実行される場合もあれば、失敗する場合もあります。あるいは、まったく実行されない場合もあります。これにより、システム開発に大きな困難が生じます。実際、さまざまな分散プロトコルは、さまざまな異常な状況下でもシステムが正常に動作することを保証するためのものです。したがって、分散システムについて学習するときは、ドキュメントのフォールト トレランスの章に注意する必要があります。
• CAP 理論: これは分散システムを研究する際に理解しておくべき重要な理論です。この理論は建築設計にも応用できます。たとえば、一貫性を低下させることでシステムの可用性を向上できる場合もあります。各データベース更新操作をバッチ操作に変換するのが典型的な例です。

CAP 理論では、3 つの文字がシステムの 3 つの矛盾する特性を表します。

• C (一貫性): 強力な一貫性。データ内のデータが完全に一貫していることを保証します。
• A（利用可能）：システムに異常があっても、サービスは提供可能です。注: ここでの可用性には、システムが正常に実行され、結果を返すことができ、また応答速度が一定に保証されていることが必要です。
• P (ネットワークの分断に対する耐性): 分散システムであるため、多くのコンポーネントが異なるサーバーに配置され、ネットワーク通信を通じて連携して動作します。これには、一部のノード サーバーでネットワーク パーティションの異常が発生した場合でも、システムが正常に動作できることが必要です。

CAP 理論によれば、CAP プロパティを完全に備えた分散プロトコルを設計することは不可能です。

上記の CAP の概念から、テクノロジを選択する際には、ニーズに基づいて、AP の高可用性 (不整合なデータ戻りを許容) を備えたシステムが必要か、CP の強力な一貫性を備えたシステムが必要かを決定するか、システムによって提供されるパラメータに基づいて 2 つを比較検討する必要があるという結論を導き出すことができます。 （読者の中には、なぜ P が必要なのかと疑問に思う人もいるかもしれません。分散システムであるため、ネットワーク分断異常が発生した場合でも、正常にサービスを提供する必要があります。）

データストレージシステム

データ量が多すぎて単一のマシンで処理できる限界を超える場合は、分散データ ストレージ システムが必要になります。オープンソース システムを選択する場合でも、独自に設計する場合でも、最初に考慮すべきことはデータをどのように配布するかです。

データ配信方法

ハッシュ方式:ハッシュ方式は、データを配布する最も一般的な方法です。大きなハッシュ テーブルを想像してみてください。各バケットがストレージ サーバーに対応し、各データは何らかの方法でハッシュ値を計算して対応するバケットに割り当てられます。 int serverId = data.hashcode % serverTotalNum 上記は単なる簡単な計算式の例です。このようにして、データをさまざまなサーバーに分散することができます。

• 利点: データとサーバー間のマッピングに関するメタデータを保存する必要がなく、serverId とサーバー IP 間のマッピングのみを記録する必要があります。
• デメリット: スケーラビリティが低い。クラスターを拡張する必要がある場合は、クラスター内のすべてのデータを移行する必要があります。最良のシナリオでも、クラスターが飛躍的に拡張された場合でも、クラスター内のデータの半分は移行する必要があります (時間があるときにこの質問について考えてください。なぜ半分だけ移行すればよいのでしょうか?)。もう 1 つの問題は、何らかのハッシュ計算の後、データが特定のサーバーに送られ、データの偏りが発生することです。
• アプリケーション例: ElasticSearch データ分散はハッシュ方式であり、routingId モジュロに基づいてデータを異なるノードにマッピングします。

データ範囲の分布:データの特定の特性値を、値の範囲に応じて異なる間隔に分割します。たとえば、時間または間隔によって、異なる時間範囲を異なるサーバーに割り当てることができます。

• 利点: データ間隔を自由に分割できます。データスキューが発生した場合、つまり、特定の間隔内のデータ量が非常に大きい場合は、間隔を分割してデータを再配布することができます。クラスターは簡単に拡張できます。新しいノードを追加する場合は、大量のデータを持つノードのデータを新しいノードに移行するだけで済みます。
• デメリット: 大量のメタデータ (データ間隔とサーバー間の対応) を保存する必要があります。
• アプリケーション例: Hbase のデータ分散では、データの行キーを使用して間隔を異なるリージョン サーバーに分割し、リージョン分割をサポートします。

データ量の配分:データ量の配分に関しては、簡単な例を考えてみましょう。ログ ファイルを使用してシステム操作のログ情報を記録する場合、ログ ファイルが一定のサイズに達すると、新しいファイルが生成され、後続のログ情報の記録が開始されます。この保存方法は、データの特性タイプとは関係ありません。これは、大きなファイルを固定サイズの複数のブロックに分割することを意味します。

• 利点: データ スキューの問題がなく、データ移行が非常に高速です (ファイルは複数のブロックで構成され、ブロックは異なるサーバー上にあるため。ファイルを移行する場合、複数のサーバーがこれらのブロックを並行してコピーできます)。
• デメリット: 大量のメタ情報 (ファイルとブロックの対応、ブロックとサーバーの対応) を保存する必要があります。
• アプリケーション例: HDFS ファイル ストレージはデータ ブロックごとに分散されます。

一貫性のあるハッシュ:前述のように、ノードを追加または削除すると、すべてのノードがデータ移行に参加し、クラスター全体が影響を受けます。そうすると、一貫性のあるハッシュによってこの問題をうまく解決できます。コンシステント・ハッシュのデータ分散方法はハッシュとほぼ同じです。唯一の違いは、コンシステントハッシュの値の範囲がリングであることです。

• 利点: クラスターは優れたスケーラビリティを備えています。ノードを追加または削除すると、隣接するデータ ノードのみが影響を受けます。
• デメリット: 上記のメリットはデメリットでもあります。ノードに障害が発生すると、すべての負荷が隣接ノードに伝達され、隣接ノードに負担がかかる可能性があります。
• アプリケーション例: Cassandra データ分散ではコンシステント ハッシュが使用されますが、コンシステント ハッシュの改良版である仮想ノードのコンシステント ハッシュが使用されます (興味のある方は学習できます)。

データ分散の問題について議論した後、次のステップは、特定のノード サービスに到達できない場合でもシステムが正常に動作し続けるという問題 (分散システム CAP におけるネットワーク パーティション異常問題) をどのように解決するかを考えることです。この問題の解決策は非常に簡単で、複数のコピーによってデータのストレージを増やし、それらを異なるノードに分散することです。ノードがハングアップすると、他のデータ コピーから読み取ることができます。

複数のレプリカを導入すると、一連の疑問が生じます。複数のレプリカのうち、どのレプリカのデータが読み取り時の基準として使用されるのでしょうか?アップデートが成功したとみなされるのは何ですか?すべてのレプリカが正常に更新された場合にのみ更新が成功したと見なされますか、それとも一部のレプリカが正常に更新された場合にのみ更新が成功したと見なされますか?これらの疑問は、実際には CAP 理論における可用性と一貫性の問題です。プライマリ-セカンダリ コピー制御モデルは、この種の問題を解決する効果的な方法です。

一次二次制御モデル

プライマリ/セカンダリ モデルは、一般的なレプリカ更新および読み取りモデルです。このモデルは比較的単純です。レプリカ関連の制御はすべて中央ノードによって制御され、同時データ変更もプライマリノードによって制御されます。この方法では、問題を単一マシンの問題に簡略化することができ、システムの複雑さが大幅に簡素化されます。

注: 一般的に使用されるレプリカ更新読み取りアーキテクチャには、プライマリ/セカンダリと分散型の 2 つがあります。プライマリ-セカンダリ構造の方が一般的ですが、分散型構造では、paxos、raft、vector time などのプロトコルがよく使用されます。私の能力が限られているため、ここでは説明しません。興味があれば自分で勉強してみて下さい。ぜひ追加してください。

次のマスター スレーブ コピー操作が関係します。

レプリカの更新

レプリカ更新の基本的なプロセス: データ更新操作はプライマリ ノードに送信され、プライマリ ノードはデータ更新操作を他のセカンダリ レプリカに同期し、他のレプリカの同期結果に基づいてクライアントに応答を返します。さまざまなデータストレージ分散システムのレプリカ更新操作プロセスは一般的に同じです。唯一の違いは、プライマリ レプリカの更新操作が完了した後にクライアントに応答するタイミングであり、これはシステムの可用性と一貫性の要件に密接に関連しています。

MySQL のマスターとスレーブを例に挙げてみましょう。通常の状況では、MySQL の更新では、クライアントに応答するためにマスターが正常に更新されるだけでよく、スレーブは binlog を通じてゆっくりと同期できます。この場合、スレーブの読み取りに一定の遅延が発生し、一貫性は比較的弱くなりますが、システムの可用性は保証されます。別のスレーブ更新戦略では、データ更新操作ではマスターが正常に更新されるだけでなく、スレーブも正常に更新される必要があります。プライマリ データとセカンダリ データは同期されたままであり、システムは強力な一貫性を保証しますが、可用性は比較的低く、応答時間は長くなります。

上記の例では、レプリカは 2 つだけです。強力な一貫性が必要な場合は、すべてのレプリカが更新された場合にのみ更新が成功したと見なされます。応答時間は比較的許容範囲内です。しかし、レプリカがさらに多い場合、一定の可用性を満たしながら一定の一貫性を確保する方法はあるのでしょうか?ここで、Quorum プロトコルについて検討する必要がありますが、その理論は簡単な数学の問題で説明できます。

N 個のレプリカがあり、更新中にそのうち W 個が正常に更新されます。次に、R レプリカを読み取ります。どのような条件下で、W と R は、読み取る R レプリカの 1 つが最新のデータであることを保証するのでしょうか (各レプリカにバージョン番号があり、バージョン番号が大きい方が最新のデータであると仮定)?

質問の答えは、W+R > N です (興味があれば考えてみてください)

クォーラム プロトコルを使用すると、ある程度の可用性と一貫性を確保しながら、レプリカ更新の成功回数をレプリカの総数の半分 (つまり、N/2+1) に設定するのが最もコスト効率が高くなります。 (Zookeeper サーバーの最適な数がカーディナリティである理由がわかりましたか?)

レプリカの読み取り

レプリカの読み取り戦略は、一貫性の選択に関連しています。強力な一貫性が必要な場合は、プライマリレプリカからのみ読み取ることができます。最終的な一貫性が必要な場合は、セカンダリレプリカから結果を読み取ることができます。最新のデータを読み取る必要がある場合は、クォーラム プロトコルの要件に従って、対応する数のレプリカを読み取ります。

コピー間の切り替え

システム内のレプリカが使用できない場合は、後続のシステムの正常な実行を保証するために、残りのレプリカの 1 つをプライマリ レプリカとして選択する必要があります。ここでは 2 つの問題が関係しています。

• レプリカの状態を判断し、ブレイン スプリットの問題を防ぐ: 一般的な方法は、Zookeeper のセッション ノードと一時ノードを使用することです。基本原則は、リース契約と定期的なハートビートです。リース契約は、簡単に言えば、両当事者間で合意された契約であると理解できます。 Zookeeper の場合、このコミットメントは、セッションの有効期間内に、ノードのステータスが有効で利用可能であると信じるということです。セッション タイムアウトが発生した場合、誤判断であるか、実際にサービスがダウンしているかに関係なく、レプリカが配置されているサービスは利用できなくなったと見なされます。このメカニズムにより、脳分裂の発生を防ぐことができます。しかし、これによって別の問題が発生します。セッション タイムアウト中にプライマリ レプリカ サービスがクラッシュすると、サービスが一定期間利用できなくなります。
• プライマリ レプリカの決定: この問題は、実際にはレプリカが最新のデータを読み取る場合と同じです。プライマリ レプリカは、クォーラムとグローバル バージョン番号を使用して決定できます。リーダー選出のプロセスでは、Zookeeper は実際にクォーラムとグローバル トランザクション ID (zxid) を使用してプライマリ コピーを決定します。

ストレージアーキテクチャモデル

データ分散とレプリカ モデルの詳細が詳しく説明されています。システム全体のアーキテクチャの観点から、データストレージの一般的なプロセスと主要なモジュールは何ですか?メタデータの保存とノード間のメンバーシップ管理の観点から、主に次の 2 つのカテゴリに分けられます。

集中型ノードメンバーシップ管理アーキテクチャ

このタイプのシステムは、主に 3 つのモジュールに分かれています。ユーザーとシステムの内部モジュール間の通信を担当するクライアント モジュール。メタデータの保存とノードの健全性状態の管理を担当するマスター ノード モジュール。データの保存とデータ クエリの返送に使用されるデータ ノード モジュール。

データ クエリ プロセスは通常、次の 2 つのステップに分かれます。

1. データに対応するノード情報をマスターノードに照会します。
2. 返されたノード情報に従って対応するノードを接続し、対応するデータを返します。

hdfs、hbase、Elastic Search などの現在の一般的なデータ ストレージ システムを分析し、上記の一般的なシステムと比較すると、マスター ノード モジュールは、具体的には hdfs の namenode、hbase の hMaster、Elastic Search のマスター ノードに対応していることがわかります。データ ノードは、hdfs のデータ ノード、hbase のリージョン サーバー、Elastic Search のデータ ノードに対応します。

分散型ノードメンバーシップ管理アーキテクチャ

以前のモデルと比較して、このアーキテクチャにはマスターノードが存在しないという最大の変更点があります。システム内の各ノードは、システム メタデータの保存やクラスター ノードの管理など、マスターと同様のタスクを実行できます。

データをクエリする方法も異なります。クライアントはシステム内の任意のノードにアクセスでき、マスター ノードに制限されなくなりました。具体的なクエリ プロセスは次のとおりです。1. システム内の任意のノードをクエリします。データがこのノード上にある場合は、対応するデータを返します。このノード上にない場合は、対応するデータのノード アドレスを返し、2 番目のステップを実行します。 2. データに対応するアドレスを取得した後、関連するデータを要求します。

ノード間でステータス情報をどのように共有するのでしょうか?一般的な方法は、ゴシップなどのプロトコルと、これに基づいて開発された serf フレームワークを使用することです。興味がある場合は、redis クラスターと consul の実装を参照してください。

データ計算および処理システム

一般的に使用されるデータ コンピューティングは、主にオフライン バッチ コンピューティング (リアルタイム コンピューティングまたは準リアルタイム ミニ バッチ コンピューティング) に分けられます。オープンソース システムは数多く存在し、各システムには独自の重点分野がありますが、共通する問題もいくつかあります。

データ配信戦略

データ処理で最初に考慮すべきことは、データ レコードがシステム内で何回処理されるか (正常および異常な状況を含む) です。

• 最大 1 回: データは最大 1 回処理されます。このセマンティクスにより、異常な状況でデータが失われる可能性があります。
• 少なくとも 1 回: データは少なくとも 1 回処理されます。このセマンティクスによりデータの重複が発生します。
• 正確に 1 回: データは 1 回だけ処理されます。このセマンティック サポートは最も複雑です。この目標を達成するには、データ処理のあらゆるリンクでこれを保証する必要があります。
• 正確に 1 回を達成するには、データ処理の各段階でいくつかの保証を行う必要があります。
• データ受信: さまざまなデータ ソースによって保証されます。
• データ転送: データ転送は 1 回だけであることが保証されます。
• データ出力: データ出力のタイプによって決まります。データ出力操作が同じデータ入力に対して冪等性を保証する場合、これは非常に簡単です (たとえば、Kafka オフセットを出力 MySQL ID として使用できます)。そうでない場合は、2 フェーズ コミットなどの追加の分散トランザクション メカニズムを提供する必要があります。

異常タスクの処理

データ コンピューティング ノードはステートレスであり、タスクのコピーを開始するだけで済むため、例外処理はデータ ストレージ システムよりもはるかに簡単です。

注: 失敗したタスクと時間制限のあるタスクに加えて、異常なタスクには、特別なタイプのタスクである遅れタスクも含まれます。大きなジョブは複数の小さなタスクに分割され、同時に実行されます。タスクの実行速度は、同じタイプの他のタスクよりもはるかに遅いことがわかります (データ スキューによって発生する実行速度の低下要因は無視します)。

タスク回復戦略は次のとおりです。

• 単純なブルートフォースで、タスクを再起動して関連データを再計算します。代表的な用途: Storm。特定のデータ実行がタイムアウトまたは失敗すると、ソースからのトポロジでデータが再計算されます。
• チェックポイントに基づいて失敗したタスクを再試行します。代表的なアプリケーション: MapReduce。完全なデータ処理は複数の段階で完了します。各ステージ (map または Reduce) の出力結果は、対応するストレージに保存されます。タスクを最初から再実行せずに実行を継続するには、タスクを再起動して前のステージの出力結果を再度読み取るだけです。

バックプレッシャー

データ処理では、上流のデータ処理がデータを消費する速度が速すぎて、MySQL などの下流のデータ出力端末に負担がかかってしまうという問題がよく懸念されます。通常の解決策は、オンラインになる前に詳細なテストを行って、下流のデータ システムが耐えられる最大の圧力を評価し、次に、1 秒あたりに消費されるデータの最大量を制限するなど、上流データのフロー制御を構成することです。実際、これはよくある問題です。現在、スパークストリーミング、ストームなど、さまざまなリアルタイムデータ処理システムがバックプレッシャー機能を提供しています。下流のデータ処理速度が遅すぎる場合、システムは上流のデータの消費速度を自動的に低下させます。

バックプレッシャーに興味がある方、データ処理システムを自分で実装したい方は、Reactive Stream を参考にしてください。このプロジェクトは、一般的なデータ処理の仕様を提供します。 Akka はこの仕様を採用している有名なプロジェクトです。

データ処理の一般的なアーキテクチャ

データ処理のアーキテクチャは一般的に似ており、通常は次のモジュールが含まれます。

• クライアント: コンピューティング タスクの送信を担当します。
• スケジューラ: コンピューティング タスクを生成し、コンピューティング リソースをスケジュールします。また、コンピューティング タスクの実行状態を監視し、異常なタスクを再起動します。
• ワーカー: コンピューティング タスクは多数の小さなタスクに分割されます。ワーカーはこれらの小さなタスクの実行を担当し、現在のノードで利用可能なリソースとタスクの実行ステータスをスケジューラに報告します。

上の図は一般的なアーキテクチャモデル図です。これは Hadoop v1 バージョンの MapReduce コンピューティング フレームワーク ダイアグラムなのかと疑問に思う人もいるかもしれません。これで、yarn モードの新しいコンピューティング フレームワーク ダイアグラムができました。このモデルをまだ使っている人はいますか?ハハ、その通りです。しかし、このモデルを使用する処理フレームワークはまだいくつかあります - Storm。

図のいくつかの概念を Storm の概念にマッピングしてみましょう。ジョブ トラッカーは Nimbus に対応し、タスク トラッカーはスーパーバイザーに対応し、各スーパーバイザーもワーカー スロットで構成する必要があり、ワーカーは Storm のワーカーに対応します。こうして比べてみると、同じように見えませんか？

このフレームワーク モデルには問題があります。責任が明確ではなく、各モジュールが異なるタスクを実行します。たとえば、ジョブ トラッカーはタスクの実行ステータスを監視するだけでなく、タスクのスケジュールも担当します。 TaskTracker は、タスクのステータスと実行だけでなく、ノード リソースの使用状況も監視します。

上記の問題に対処するために、YARN モードに基づく新しい処理アーキテクチャ モデルでは、タスク実行ステータスの監視とタスク リソースのスケジュールを分離します。元のジョブ トラッカーは、リソースのスケジュールを管理するリソース マネージャーと、タスクの実行を監視する各 appMaster に分かれています。元のタスク トラッカーはノード マネージャーになり、リソースの監視とタスクの開始を担当し、タスクの実行ステータスと例外処理は appMaster によって処理されます。

同様に、Twitter は Storm アーキテクチャのいくつかの問題に基づいて、新しい処理フレームワーク Heron を立ち上げました。これが解決する問題は、タスクのスケジュールとタスク実行ステータスの監視の責任を分離することであり、ここでは AppMaster に似た新しい概念である Topology Master を導入します。

要約する

分散システムは多くのコンテンツをカバーします。この記事では、主に全体的なアーキテクチャと概念から開始する方法を紹介します。学習プロセスにはいくつかの一般的な問題があり、以下にまとめます。

• まず、システムがデータ ストレージ システムなのかコンピューティング システムなのかを分析します。
• データ ストレージ システムの場合は、データの分散とレプリケーションの戦略から始めます。データ処理の問題である場合は、データ配信戦略から始めます。
• よく使われるアーキテクチャモデルに対応するシステムアーキテクチャ図を読み、各コンポーネントを既存のシステムと比較し、このコンポーネントを HDFS のネームノードに類似したものと考えるなどして、最終的にデータフローのプロセス全体を頭の中で整理します。
• システムの全体的な状況を理解した後、ドキュメントのフォールト トレランス セクションに特に注意して、システムがどのように障害を許容するかを確認します。あるいは、事前に自分自身にいくつか質問をすることもできます。たとえば、ノードまたはタスクに障害が発生した場合、システムはこれらの例外をどのように処理するのでしょうか?疑問を念頭に置いて文書を読んでください。
• ドキュメントを詳しく読んだ後、公式ドキュメントに従っていくつかの hello world の例を記述し、システム構成項目を詳細に確認できます。作業が進むにつれて、システムの詳細や主要なソースコードを確認できるようになります。

今回シェアさせていただいた記事の内容は以上です。そこには必然的にいくつかの省略があります。ご質問がございましたら、いつでもお気軽にご指摘、ご連絡ください。一緒に進歩しましょう。ありがとう。

著者: Li Feng、シニアエンジニア。現在はLogicMonitor（SaaSサービス監視プラットフォームを提供し、毎日数百億の監視データを収集）に勤務し、分散ストレージストリーミングコンピューティングを中心としたデータ処理プラットフォームのアーキテクチャに従事しています。クリックすると元のテキストが表示され、やり取りの記録が表示されます。