EMR creates efficient cloud native data analysis engine

オープンソースシステムに基づくクラウド上のデータ分析プラットフォームの構築

お客様がオープンソースソリューションを選択する主な理由は以下のとおりです。


・柔軟で多様なビジネスシナリオ：現在、中小企業であっても、ビジネスデータ、ログデータ、グラフデータなど多種多様なデータを保管しています。
このような場合、異なるビジネスシナリオをシームレスに連携させるために、高度にカスタマイズされたシステムが必要です。

・専門的な運用保守能力：オープンソースシステムには豊富な人材リソース、充実したオンライン情報、そして強力なオープンソースコミュニティの支援があり、ビジネスの円滑な推進を保証できます。

・多様なビジネスニーズとコスト圧力のバランス：各クラウド製品にはそれぞれの適用シーンがあります。
中小企業にとって、複数のクラウド製品を購入することは大きなコスト負担となります。
業務に必要なコンポーネントのみを選定することで、ユーザーのコストを削減できます。


下図は Alibaba EMR システムの製品アーキテクチャ図です。
クラウドへの移行には主に 2 つの方法があります。
1 つは ECS リソースを購入してオープンソースシステムを自社で構築する方法、もう 1 つは Alibaba の EMR システムを直接選択する方法です。
前者の方法では、Spark、Hive、Flink、TensorFlow など多くのオープンソースコンポーネントが関係します。
ユーザーがゼロから完全なビッグデータシステムを構築することは非常に複雑であり、特に数百・数千規模のクラスターの場合、運用保守担当者にとっても大きな課題となります。
EMR システムの利用には以下のメリットがあります。


1) Alibaba Cloud EMR システムは、ワンクリックで関連コンポーネントのデプロイと設定を自動的に行い、すぐに使える状態にします。
同時に、ユーザーのマシンタイプに応じてパラメータを自動で推奨・調整します。


2) Alibaba Cloud EMR システムは他の Alibaba Cloud 製品と連携しています。
たとえば、データは OSS に保存され、EMR システムは追加の認証設定なしで OSS 上のデータを簡単に読み取れます。


3) Alibaba Cloud EMR システムには、他の製品にはない独自開発のプラグインが多数組み込まれています。


4) Alibaba Cloud EMR システムのすべてのコンポーネントはオープンソースと互換性がありながら、それを超える性能を備えています。
たとえば、Flink には Alibaba Cloud 独自開発の Blink と TensorFlow (PAI) が統合されており、ユーザーは Alibaba Cloud の内部技術を活用できます。


5) Alibaba Cloud EMR システムは、プラットフォーム全体のジョブ診断・アラートコンポーネント APM を提供し、自動化された運用保守を実現することで、クラスター運用保守の複雑さを大幅に低減します。


6) Alibaba Cloud EMR システムは DataWorks とも連携しており、ユーザーは DataWorks をエントリーポイントとして EMR システムを簡単に利用できます。


EMR システムの主な目標は 3 つあります。


・プラットフォーム化：EMR をクラウド上の統一されたデータ分析プラットフォームとし、フルスタックのビッグデータソリューションの構築を支援します。
VM の完全なコンテナ化をサポートし、エンタープライズグレードの HAS とビッグデータ APM を提供します。


・技術コミュニティと深耕：技術コミュニティへの貢献を継続し、ビッグデータに優しいクラウドネイティブストレージを構築するとともに、技術をコミュニティに還元し、貢献を行います。


・エコシステム：EMR システムは他の Alibaba Cloud 製品と組み合わせてエコシステムを構築し、Blink や PAI と連携し、OSS や OTS のソリューションを統合します。


EMR-Jindo：クラウドネイティブな高効率データ分析エンジン

下図は TPC-DS ベンチマークテストレポートです。
2019 年 3 月の 10TB テストでは、パフォーマンスインデックススコアは約 182 万で、コストは 0.31 USD でした。
一方、2019 年 10 月の同テストでは、パフォーマンスインデックススコアが 526 万に向上し、コストは 0.53 CNY に低下しました。
つまり、約半年でパフォーマンスが 2.9 倍に向上し、コストは元の 4 分の 1 に削減されたことになります。
同時に、Alibaba は TPC-DS テストの 100TB テストレポートを提出した最初のメーカーとなりました。
これらの成果の背景には、EMR-Jindo エンジンの支えがあります。


EMR-Jindo エンジンアーキテクチャは主に 2 つの部分に分かれます。


・Jindo-Spark：EMR 内部の完全に最適化された Spark 高効率コンピューティングエンジンで、多様なコンピューティングタスクを処理できます。


・Jindo-FS：独自開発のクラウドネイティブストレージエンジンで、オープンソースの HDFS インターフェースと互換性があり、パフォーマンスとコストの両立を図っています。


1) Jindo-Spark
Jindo-Spark 高効率コンピューティングエンジンは、Spark に対して一連の最適化措置を採用しています。
Runtime Filter による適応型ランタイムデータクリッピングのサポート、Enhanced Join Reorder による外部結合の並べ替えなどの課題解決、TopK の推論と TopK ロジックのプッシュダウンによる正確なデータフィルタリング、File Index によるファイルレベルのフィルタリング（min/max/bloom/inverting など）、独自開発の Relational Cache によるクエリの分鐘レベルからサブ秒レベルへの高速化などがあります。
さらに、特定のシナリオ向けに Spark Transaction 機能を導入してフル ACID サポートを実現し、Smart Shuffle 機能によってソートマージの回数を基盤層から削減して Shuffle の効率を向上させています。


・Runtime Filter：
Spark の Dynamic Partition Pruning (DPP) に似ていますが、DPP よりも強力です。
DPP が処理できる分析テーブルに加え、Runtime Filter は非分析テーブルも処理できます。
基本原理は、ランタイム時にデータを動的にトリミングし、不要な計算を回避することです。
たとえば、結合クエリにおいて、値をストレージレイヤーにプッシュダウンしてフィルタリングできない場合や、論理計算時に最終的なデータ量を予測できない場合、分析テーブルであれば、Runtime Filter はまず一方のテーブルの結合操作に関与するデータ量を推定します。
データ量が少ない場合は、事前にデータをフィルタリングしてからもう一方にプッシュしてフィルタリングを行います。
非分析テーブルには BloomFilter などの Filter を導入して Min や Max の統計情報を取得し、

これらの統計情報に基づいて、候補データが少ない側を抽出してもう一方にプッシュしてフィルタリングします。
Runtime Filter のオーバーヘッドは非常に小さく、オプティマイザー内での評価のみで済み、大幅なパフォーマンス向上をもたらします。
下図に示すように、Runtime Filter は全体で約 35% のパフォーマンス向上を達成しています。
この機能は Spark に PR として提出されています (SPARK-27227)。


・Enhanced Join Recorder：
ご存知のとおり、演算子の実行順序は SQL の実行効率に大きな影響を与える可能性があります。
この場合、最適化の核心原則は、演算子の実行順序を変更し、できるだけ早い段階でデータをフィルタリングすることです。


たとえば、下図の左上の例では、下部の 2 つのテーブルが非常に大きい場合、これらを結合するオーバーヘッドは非常に高くなります。
結合後の大量データがさらに小さなテーブルと結合され、ビッグデータが層ごとに転送されます。
これが続くと、プロセス全体の実行効率に影響します。
このとき、最適化の考え方は、大きなテーブルから無関係なデータを先にフィルタリングし、下流に転送されるデータ量を削減することです。
この問題を解決するため、Spark は動的計画法を使用していますが、テーブル数が比較的少ない場合にのみ適用可能です。
テーブル数が 12 を超えると、このアルゴリズムは対応できません。
多数のテーブルの場合、EMR は多テーブル結合の遺伝的アルゴリズムを提供し、元の動的計画法の 2n の複雑さを線形レベルに低減し、数百・数千のテーブルの結合を完了できます。


下図の右上を見ると、Query64 では 18 のテーブルが結合に参加しており、動的計画法の最適化時間は 1400 秒かかるのに対し、多テーブル結合の遺伝的アルゴリズムは約 20 秒で完了します。
Join Recorder のもう一つの重要な機能は、外部結合の並べ替えアルゴリズムです。
ご存知のとおり、SQL の外部結合の順序は任意に交換できませんが、交換できないわけではありません。
たとえば、A left join B、さらに left join C の場合、実は特定の条件下では順序を交換できます。
Spark では外部結合の最適化は直接放棄されていますが、EMR は既存の研究に基づいて順序交換の十分必要条件を見つけ出し、外部結合の並べ替えアルゴリズムを実装しました（下図の左下）。
外部結合の実行効率は質的に向上しました（下図の右下）。


・Relational Cache：
Spark の元の Cache にはいくつかの制限があります。
まず、Spark の Cache はセッションレベルです。
特定のクエリフラグメントが頻繁に使用されていることが判明した場合、そのセッションにキャッシュが作成されますが、セッション終了後にキャッシュは消滅します。
次に、Spark の Cache はローカルに保存され、分散ではないため、汎用的に使用できません。
これに対し、EMR プラットフォームは Relational Cache を実装しました。
Spark テーブル、ビュー、Dataset などの関係データの任意の抽象データエンティティにキャッシュを作成します。
マテリアライズドビューに似ていますが、より多くの機能を備えています。
Relational Cache の利用シナリオには、a) サブ秒レスポンスの MOLAP エンジン、b) インタラクティブ BI、Dashboard、c) データ同期、d) データ事前整理があります。


Relational Cache の作成プロセスは以下のとおりで、構文は Spark SQL の一般的な DDL に似ています。
まず、テーブルまたはビューをキャッシュし、次に Relational Cache の更新戦略（DEMAND または COMMIT）、後続の最適化に使用するかどうか、Cache データの保存方法、Cache のビューロジックを指定します。
Relational Cache は任意の Table と View のキャッシュをサポートし、メモリ、HDFS、OSS など任意のデータソースにキャッシュでき、JSON、ORC、Parquet など任意のデータ形式に対応します。


Relational Cache は、ユーザーが入力する SQL の最適化もサポートします。
元の Spark SQL Cache はユーザー入力の SQL 最適化において非常に硬直的で、ユーザー入力の SQL が Cache と完全に一致しなければ使用できませんでした。Relational Cache は全く異なり、たとえばテーブル a、b、c、d の 4 テーブル結合のキャッシュがある場合、テーブル a、b、e の 3 テーブル結合があるとき、a と b の結合結果を 4 テーブル結合のキャッシュデータから読み取れます。下図の右側は、Cache の有無によるベンチマークテスト結果を示しています。Relational Cache はテストのレスポンス時間をサブ秒レベルで保証できることがわかります。
サブ秒レスポンスのインタラクティブ分析を実現する Spark Relational Cache をご参照ください。

・Spark Transaction：一部のユーザーは Hive テーブルを使用している場合があります。Hive テーブルにはトランザクションサポートがありますが、Spark はトランザクションに関して Hive と互換性がありません。そのため、ユーザーのデータ修正/削除およびデータストリームインポートのシナリオに対応するため、EMR プラットフォームは Spark Transaction の ACID サポートを提供します。

従来のデータインポートはバッチで行われ、たとえば 1 日 1 回などですが、ストリーミングデータインポートのシナリオでは、データは一切の処理なしでリアルタイムに書き込まれる生データであるため、削除と更新のニーズが生じます。一般的に、Spark Transaction はロック + MVCC の実装であり、MVCC は基盤ストレージと切り離せません。Hive と Spark の互換性において、ビッグデータはファイル形式でディレクトリに保存されます。ファイルのバージョンは行単位で管理され、書き込まれた各行にはメタカラム（op、original_write-id、bucket id、row_id など）が追加され、テーブル全体で行を一意に識別します。行の更新が必要な場合、その行をインプレースで更新するのではなく、行を取り出して書き直し、新バージョンを生成して保存します。読み取り時に、複数のバージョンがマージされてユーザーに返されます。

###2) Jindo-FS
EMR は初期にローカルディスクモデルをリリースしました。このモデルでクラスターをデプロイすると、クラウド下でローカルクラスターを使用してビッグデータディストリビューションをデプロイするのに似ており、価格は比較的高めです。さらに、当時 HDFS にはメタデータのボトルネックがあり、ローカルストレージのスケーラビリティは大きな課題に直面していました。この問題に対するソリューションは、コンピューティングとストレージを分離し、データを OSS に保存することです。しかし、この分離の直接的な結果としてパフォーマンスが低下します。OSS のメタデータ操作は時間がかかり、データの読み取りはネットワークを越えるため、転送帯域幅もパフォーマンスに深刻な影響を与えます。

さらなるソリューションは、リモートからデータをコンピューティング側にプルしてキャッシュすることです。これが Jindo-FS の役割です。Jindo-FS は HDFS に似たシステムで、アーキテクチャも HDFS のマスター・スレーブアーキテクチャに似ており、Name Service と Storage Service に分かれています。アクセス頻度の高い一部のテーブルを RocksDB に配置してマルチレベルキャッシュをサポートします。Jindo-FS 全体は HDFS のマスターノードとは異なり、Jindo-FS の「マスター」（Name Service）は Raft プロトコルを使用してサービスを提供する分散クラスターです。複数のネームスペースをサポートし、メタデータは kv 形式で高性能 kv ストアに保存されます。データ自体は保存せず、実際のデータは OSS と OTS にあるため、弾力的なデータ拡張と破棄再構築をサポートします。

Jindo-FS の基盤層のメタデータ管理は、データを kv のシリーズに分割し、増分 ID を通じて階層ごとにクエリします。たとえば、/home/Hadoop/file1.txt は OTS を 3 回読み取る必要があります。下図の右側のテスト結果は、Jindo-FS がメタデータ操作において OSS よりも優れたパフォーマンス向上を示すことを示しています。

Jindo-FS は基盤ストレージに Storage Service を使用します。書き込みプロセスでは、Storage Service は書き込むファイルをローカルと OSS に同時に保存し、書き込み結果をユーザーに返します。同時に、クラスターノード間でマルチコピー転送も実行します。一方、読み取り操作は HDFS に似ており、ローカルにヒットした場合はローカルストレージから読み取り、そうでない場合はリモートから読み取ります。Storage Service は高性能、高信頼性、高可用性、弾力ストレージの特徴を備えています。高性能をサポートするため、Jindo-FS は高速データフローチャネルを構築し、メモリコピー回数の削減など一連の戦略も採用しています。