Use E-MapReduce to build a data lake on the cloud

データレイク

データレイクは 15 年前に提唱され、この 2〜3 年で非常に人気を高めています。Gartner のマジッククアドラントにおいて、データレイクは大きな投資と探索の価値がある技術とされています。

データレイクとは何でしょうか。以前はデータウェアハウスを使用して構造化データを管理していました。Hadoop の台頭後、大量の非構造化データと構造化データが HDFS に統一して保存されるようになりました。しかし、データの蓄積に伴い、一部のデータには収集に適したアプリケーションシナリオがない場合があり、まずは保存しておき、ビジネス上の必要が生じた際にさらに開発やマイニングを行うことがあります。

データ量の継続的な増加に伴い、OSS などのオブジェクトストレージや HDFS を使用して統一ストレージを実現できます。同時に、アドホッククエリ、オフラインコンピューティング、リアルタイムコンピューティング、機械学習、ディープラーニングなど、さまざまな計算シナリオに対応する必要があります。異なる計算シナリオでは、異なるエンジンの選択肢に直面します。また、どのシナリオにおいても、監視、権限管理、監査、アカウントシステムを統一して実装する必要があります。

最初の部分はデータ取得です (図の最も左側のフレーム)。主にリレーショナルデータベースの収集に使用されます。ログやユーザーのクリックストリームを統一ストレージに保存し、異なる計算サービスを使用してデータを処理・計算し、計算結果を AI 分析プラットフォームに適用して機械学習やディープラーニングを行い、最終的にその結果をビジネスに活用し、検索やソースデータ管理などの機能を使用してデータの付加価値効果を実現します。計算とストレージに加え、データには一連の管理と監査の手段も必要です。

ビッグデータ技術の誕生から 10 年以上が経過しました。当初、人々は自社 IDC にオープンソースソフトウェアを構築していました。ビジネスの継続的な成長、データの急速な蓄積、ビジネスの急激な変動に伴い、突発的なビジネス急増が発生する可能性があります。

オンプレミスの IDC では調達サイクルが非常に長く、ビジネスの急速な成長に伴うコンピューティングリソースの需要に対応することが困難です。同時に、ビジネスにはピークと谷があります。日中の計算タスクは比較的少ない可能性があり (ほとんどがアドホッククエリ)、夜間にはオフラインレポート計算のためにリソースを追加する必要がある場合があります。このような場合、IDC モデルではコンピューティング能力を合わせることが困難です。

約 5〜6 年前から、多くの企業がクラウドへの移行を開始しています。ビジネスデータが継続的に増加する中、企業はクラウドのサプライチェーン機能を活用して、ビジネスの成長ニーズに対応するためにインスタンスを迅速に追加できます。クラウド上で独自の Hadoop クラスターや EMR を構築する場合にも課題があります。HDFS を本質的に使用するため、データ量の増加に伴いストレージコストが線形に増加します。同時に、クラウド上でローカルディスクを使用する場合、運用管理プロセスが非常に複雑になります。

大規模クラスター (数百台から数千台のクラスター) では、ディスク故障は日常的な出来事です。この日常的な出来事にどのように対処するかも、非常に困難な課題です。そのため、OSS を中心としたデータレイクアーキテクチャに徐々に進化してきました。OSS の階層型ストレージ機能により、異なるデータに対して異なるストレージ方法とコストを実現できます。同時に、HA シナリオ下での HDFS の NameNode の運用管理は非常に複雑です。クラスター規模が 100 を超えると、NameNode をいかに安定させるかが非常に困難な課題となり、多くのエネルギーと人的リソースを維持に費やす必要があります。HA アーキテクチャの維持は、長期的には解決が困難な問題となる可能性があります。OSS を使用することはもう一つの選択肢です。クラウドサービスのストレージアーキテクチャを使用することで、HDFS アーキテクチャにおける未解決の問題を回避できます。

EMR を使用したエンタープライズレベルのデータレイクサービスの構築

EMR は Alibaba Cloud のエコシステムを活用し、100% オープンソースであり、エンタープライズに安定的で信頼性の高いオープンソースビッグデータサービスを提供するよう位置付けられています。EMR は 2016 年 6 月にリリースされ、EMR 4.4 まで継続的にバージョンアップを重ねてきました。EMR を基盤として、ユーザーは 10 台以上の ECS インスタンスを選択して数分で弾力クラスターを作成できます。

EMR は Alibaba Cloud OSS をサポートし、独自開発の JindoFS が OSS のパフォーマンスを大幅に向上させます。同時に、EMR は Alibaba Cloud エコシステムとも統合されています。DataWorks や PAI は EMR 上でシームレスに連携できます。また、ストレージプロダクト (Log Service や MaxCompute など) に対して、EMR を計算エンジンとして内部に保存されたデータを計算できます。すべての EMR コンポーネントは Apache オープンソースバージョンです。コミュニティバージョンの継続的なアップグレードと反復進化に伴い、EMR チームは Spark、Hadoop、Kafka などのコンポーネントについて、アプリケーションとパフォーマンスの一連の最適化と改善を行っています。

EMR はセミマネージドアーキテクチャを採用しています。このアーキテクチャの下で、ユーザーはオンプレミス IDC と非常に近い体験を得られます。ユーザーはクラスター内の ECS サービスノードに実際にログインして、独自の ECS サーバーをデプロイおよび管理できます。同時に、APM のようなホスト、ジョブ、サービスレベルでの警告や診断を含む一連のエンタープライズレベルの機能を提供し、認証プラットフォームとして MIT、Kerberos、RAM、HAS をサポートし、Ranger を統一権限管理プラットフォームとして使用しています。

以下の図は EMR のオープンソースビッグデータエコシステム全体を示しており、いくつかのソフトウェアとハードウェアを含んでいます。

いくつかのレベルがあります。

たとえば、JindoFS はストレージレイヤー (OSS) 上にあります。JindoFS は EMR チームが開発したコンポーネントセットです。このコンポーネントは主に OSS データの読み取りと計算の高速化に使用されます。実際の比較テストによると、JindoFS のパフォーマンスはオフライン HDFS サービスを大幅に上回っています。

Delta Lake は Databricks のオープンソースデータレイクの技術計算エンジンおよびプラットフォームです。EMR チームは Delta Lake を中心に、Presto、Kudu、Hive との連携について一連の最適化を実施し、オープンソースバージョンと比較してパフォーマンスも大幅に向上させています。特に注目すべきは、EMR の Flink は Ververica のエンタープライズバージョンであり、パフォーマンス、管理性、保守性の面で優れた性能を発揮します。

EMR は主に 4 つのノードタイプ (Master、Core、Task、Gateway) に分類されます。

Master ノードは主に NameNode、ResourceManager、Hbase の HMaster などのサービスをデプロイします。これらのサービスは集中統一クラスター管理を実現します。本番クラスター作成時に HA を有効にすると、高可用性クラスターが自動的に作成されます。

Core ノードは主に Yarn の NodeManager と HDFS の DataNode をデプロイします。この点から、計算とストレージの両方を実行できます。ノードに保存されるデータの信頼性を考慮し、このノードでは弾力的スケーリングやプリエンプティブインスタンスを使用できません。

Task ノードは NodeManager のみをデプロイし、データレイクシナリオで弾力的にスケーリングできます。すべてのユーザーデータがオブジェクトストレージに統一して保存される場合、ユーザーは TASK ノードの弾力的なスケーラビリティを活用してビジネスの変化に迅速に対応し、コンピューティングリソースの弾力的な拡張を実現できます。同時に、ECS プリエンプティブインスタンスを使用してコストを削減できます。Task ノードは GPU インスタンスもサポートしています。多くの機械学習やディープラーニングシナリオでは、計算サイクルが非常に短く (数日または数週間に 1 回のみ計算)、GPU インスタンスは高価なため、手動でリサイズすることでコストを大幅に削減できます。

Gateway ノードは主に Spark、Hive、Flink などのさまざまなクライアントコンポーネントをデプロイするために使用されます。これにより、異なる部門が異なるクライアントやクライアント設定を使用して完全な分離を実現でき、同時にユーザーが頻繁にクラスターにログインして操作する必要性を回避します。

JindoFS

HDFS は 10 年以上にわたり発展し、コミュニティのサポート機能は比較的成熟しており完全です。しかし、使用における短所も確認されています。たとえば、HA のアーキテクチャが複雑すぎる (HA を実装するには JournalNode、ZKFC をデプロイする必要がある)、クラスター規模が非常に大きい場合は HDFS Federation を考慮する必要があるなどです。ビジネス規模が大きい場合、DataNode のデコミッション処理の期間も非常に長くなります。ホスト障害やディスク障害によりノードをオフラインにする必要がある場合、その期間が 1〜2 日に達することもあり、DataNode のデコミッション処理を管理するために専任の人員を配置する必要さえあります。NameNode の再起動には半日かかることもあります。

OSS の利点は何でしょうか。OSS は Alibaba Cloud 上のサービス指向のオブジェクトストレージです。管理と運用保守のコストが非常に低く抑えられています。同時に、標準、低頻度、アーカイブなど、さまざまなタイプの階層型データストレージがあります。このように、OSS はユーザーの使用コストを効果的に削減できます。ユーザーは NameNode や Federation に注意を払う必要がなく (サービス指向のため)、データの信頼性も非常に優れています (11 個の 9 のデータ信頼性を提供)。そのため、多数のお客様が OSS を使用してエンタープライズデータレイクを構築していることがわかります。OSS の典型的な特徴はその開放性です。基本的に、すべてのクラウドプロダクトがバックエンドストレージとして OSS をサポートしています。

同時に、OSS にも課題があります。当初、オブジェクトストレージは主にビッグデータシナリオでビジネスシステムと連携してデータを保存するために使用されていました。OSS は汎用シーンを対象に設計されているため、ビッグデータ計算エンジン (Spark、Flink) に適応する際にパフォーマンスの問題に直面します。rename 操作が実行されると、実際には move 操作、すなわちファイルコピーが実行されます。Linux ファイルシステムのように rename 操作が高速ではありません。list 操作が実行されると、すべてのオブジェクトがリクエストされ、数が多すぎると速度が極端に遅くなります。整合性 (最終整合性の期間) も比較的長くなります。読み取りと書き込み時にデータの不整合が発生する可能性があります。

EMR が独自に開発した JindoFS は、オープンソースエコシステムに基づいています。基本的にすべての計算エンジンが JindoFS を使用して OSS の読み取り、計算、クエリを実行できます。JindoFS は OSS の利点、すなわち大量データ (EB レベル) のストレージを最大限に活用できると同時に、柔軟な特徴も発揮できます。OSS セマンティクスを使用する場合、ほぼすべての計算エンジン (他の計算プロダクトや BI レポーティングツールなど) がデータを迅速に取得でき、これは共通のインターフェイスです。

JindoFS はクラウド内でも大規模に使用されています。HDFS と OSS のデータを処理する際に、ファイルの名前変更やリストなどの操作に関するパフォーマンス問題を回避できます。

JindoFS のアーキテクチャを以下の図に示します。プライマリサービスが Namespace Service で、セカンダリサービスが Storage Service です。マスターサービスは 1 つ以上のノードにデプロイでき、スレーブサービスは各ノードにデプロイされ、クライアントサービスは各 EMR マシンにデプロイされます。データの読み取りと書き込み時には、まずスレーブサービスを通じてマスターサービスにリクエストを送信し、ファイルの位置を取得します。ファイルがローカルに存在しない場合は、OSS から取得してローカルにキャッシュされます。JindoFS は HA アーキテクチャを実装しています。その HA はローカルでは RocksDB を通じて、リモートでは OTS を通じて実装されています。そのため、JindoFS はパフォーマンスと高信頼性の両方を備えています。JindoFS は Ranger を通じた権限管理と設計も可能です。JindoFS SDK を使用すると、オンプレミスの HDFS データを OSS に容易に移行してアーカイブや使用ができます。

JindoFS は Block モードと Cache モードをサポートしています。JindoFS を Block モードで使用する場合、そのソースデータはローカルの RocksDB とリモートの OTS に配置されます。OSS の一般的なソースデータではなくなります。データ量が大きい (数百 TB 以上) 場合、Block モードのパフォーマンスがより優れますが、汎用性は比較的低くなります。ユーザーは JindoFS のソースデータを通じてのみファイルブロックの位置と詳細を取得できます。同時に、JindoFS の Block モードは、ホットデータ、コールドデータ、温度データを指定することもサポートしています。JindoFS は運用管理の複雑さを効果的に低減できます。

Cache モードはローカルストレージを使用し、セマンティクスも OSS 本来のセマンティクスを使用します (oss://bucket/path など)。Cache モードを使用する利点は汎用性が高いことです。EMR だけでなく、他の計算エンジンでも使用できます。欠点はパフォーマンスにあります。データ量が非常に大きい場合、Block モードと比較してパフォーマンスが比較的劣ります。

上記 2 つのモードは、自らのビジネス判断に基づいて選択的に使用する必要があります。

弾力的スケーリング

EMR は時間とクラスターのロード (Yarn の指標収集、ユーザーが手動で指定可能) に応じて弾力的にスケーリングできます。同時に、弾力的スケーリング実行時に複数のインスタンスタイプを選択でき、在庫不足による特定のインスタンスタイプでのジョブ失敗を回避できます。また、プリエンプティブインスタンスを使用してコストを削減することもできます。

EMR のデータレイクソリューション

以下の図に示すように、これはオフラインコンピューティングアーキテクチャです。データは Kafka、Log Service、Data Integration (DataWorks 内のデータ統合)、または Lightning Cube (Alibaba Cloud のオフライン移行プロダクト) などの方法でオブジェクトストレージに同期できます。データは EMR で直接読み取りと計算が可能です。ワークフロースケジューリングには DataWorks または Airflow を使用できます。トップレベルのデータアプリケーションには、データレポート、データダッシュボード、API などがあります。

このオフラインアーキテクチャの利点は、大規模 (EB レベル) のデータストレージをサポートできることです。OSS は EMR のビッグデータ計算エンジンとシームレスに接続でき、優れたパフォーマンスを確保できます。

OSS の機能を組み合わせることで、高パフォーマンスの読み取りを実現できます。制御レベルでは、Ranger を通じてすべてのオープンソースコンポーネントの権限管理を統一できます。

次の図は EMR を使用してアドホッククエリを実行するシナリオを示しています。このシナリオではリアルタイム計算によるデータ流入があります。リアルタイム計算は JindoFS を通じて高速化され、Presto や Impala などのオープンソース計算エンジンを通じて BI レポートやビッグデータダッシュボードに接続されます。このシナリオはリアルタイムデータウェアハウス業務でよく見られます。

EMR の新機能と特徴

お客様の事例

IDC からクラウドへの移行の典型事例

お客様のデータ規模は PB レベルです。クラウド移行後、実際のホットデータが全データの約 10% を占めていることが判明しました。オブジェクトストレージの階層型ストレージにより、コストを効果的に削減しています。クラスター規模が大きい場合、マスターの負荷が大きくなりますが、このお客様ではマスター数が多くなっています。EMR はクラスターサイズと負荷の判断に基づいてマスターを動的に拡張および縮小します。同時に、マスター数をカスタマイズしてクラスターを計算およびデプロイすることもできます。さらに、お客様は弾力的スケーリング機能を使用しています。比較的独立した Spark クラスターが AI と ETL を処理しています。この Spark クラスターは弾力的スケーリングクラスターです。ビッグデータ開発において、お客様は DataWorks を使用してビッグデータ開発を行っています。Screenshot of 5.51.28.png on August 21, 2020

データ計算の高いパフォーマンスと権限管理能力によるエンタープライズ効率の向上

多数のコンピューティングクラスターを運用しています。同時にお客様はクラスターの権限管理とソースデータ管理を統一して抽象化し、集中管理を実現しています。たとえば、Hive Meta と JindoFS Meta は複数のクラスターで共有されるソースデータです。日中のクラスター数とクラスターノード数は少ないですが、夜間のビジネスピーク時の計算能力を満たすために容量を拡張する必要があります。ワークフロースケジューリングには Airflow を使用しています。