Best practice of data lake construction based on DeltaLake

1、業務背景

Homehelp はテクノロジーを基盤としたオンライン教育企業です。現在、ツール製品の Homework Help、Homework Help Mental Arithmetic、K12 ライブ授業製品の Homework Help Live Class、素質教育製品の Deer Programming、Deer Writing、Deer Art、および Meow Machine などのインテリジェント学習ハードウェアを保有しています。また、教育研究センター、教授センター、コーチングオペレーションセンター、ビッグデータセンターなど、複数の業務システムを運営し、より質の高い教育プロダクトを継続的に提供するとともに、ユーザーに優れた学習体験と利用体験をもたらし続けています。ビッグデータセンターは基盤システムの中心として、企業レベルのデータウェアハウス構築を主に担当し、保持率、出席率、アクティブユーザー数など、各プロダクトラインにビジネス指向のデータ情報を提供して、運用意思決定の効率と品質を向上させています。

上記の図はジョブヘルプのデータセンターの概要を示しています。主に 3 つのレイヤーに分かれています。

• 第 1 層はデータプロダクトおよび機能提供レイヤー

主にテーマ別データドメインに基づいて構築されたデータツールとプロダクトで、ビジネスインテリジェンス、トレンド分析などのアプリケーションシナリオを支援します。

• 第 2 層はグローバルデータレイヤー

OneModel による統一モデリングを通じて、取り込んだデータの標準化モデリングを実施し、ビジネスドメインのテーマ別データを異なるタイムリー性のシナリオに応じて構築することで、上位レイヤーのプロダクトの効率と品質を向上させています。

• 第 3 層はデータ開発レイヤー

データ統合、タスク開発、データ品質、データサービス、データガバナンスなど、企業内のすべてのデータ開発プロジェクトを支援する一連のシステムとプラットフォームを構築しています。

本共有の主な内容は、オフラインデータウェアハウス（日次レベルおよび時間次レベル）の本番運用・利用におけるパフォーマンス問題の解決です。

2、課題とペインポイント

ジョブヘルプのオフラインデータウェアハウスは Hive を基盤としており、ODS 層から ADS 層までのデータ構築機能を提供しています。ADS テーブル生成時、データ統合を通じて OLAP システムに書き込まれ、経営者向けに BI サービスを提供します。また、DWD、DWS、ADS テーブルはアナリスト向けのオフラインデータ探索およびデータ検索サービスも提供しています。

ビジネスの段階的な発展とそれに伴うデータ量の増加により、オフラインデータウェアハウスシステムでは以下の主要な問題が顕在化しています。

• ADS テーブルの出力遅延が長期化

データ量の増加により、ODS 層から ADS 層までのフルリンク構築時間がますます長くなっています。極めて重要な ADS テーブルチェーンについては、リソースを傾斜配分することで短時間での解決が可能ですが、これは本質的にリソース配分とタスク編成のモデルであり、大規模に複製できません。そのため、他の重要な ADS テーブルのタイムリーな出力に影響を及ぼしています。たとえば、アナリストにとってデータテーブルの遅延により、T+1 テーブルの最悪ケースでは T+2 まで待たされることになります。

• 時間次テーブルの需要への対応が困難

一部のシナリオでは時間単位の出力テーブルが必要です。たとえば、時間単位のフィードバックにより運用戦略を適時に調整する必要がある活動などです。このようなシナリオでは、データ量の増加とコンピューティングクラスターのリソース逼迫により、時間次テーブルのタイムリー性を確保できないことが多くなっています。コンピューティングパフォーマンスを向上させるため、事前に十分なリソースを確保する必要があり、特に日次データの時間次計算が必要な場合、最悪ケースでコンピューティングリソースを 24 倍に拡張する必要があります。

• データ探索の遅さとデータ検索の安定性不足

データの本番運用後、アナリストによる利用が行われます。Hive への直接アクセスには数十分から数時間かかり、これは到底受け入れられません。ユーザーから不満を受けることが多く、Presto を使用して Hive テーブルのクエリを高速化しています。ただし、Presto のアーキテクチャ特性上、クエリ対象のデータテーブルが大きすぎたりロジックが複雑すぎたりすると、Presto のメモリ OOM が発生します。さらに、Hive の既存の UDF や VIEW を Presto で直接使用できないことも、アナリストの活用シーンを制限しています。

3、ソリューション

問題分析

従来の ODS 層から ADS 層までの全リンク出力が遅い場合でも、特定のテーブルへの探索アクセスが遅い場合でも、本質的には Hive 層のコンピューティングパフォーマンス不足を意味しています。上記のシナリオ分析から以下の原因が判明しました。

• リンク計算が遅い原因：Hive はインクリメンタル更新をサポートしていません。一方、ビジネス層のデータソースからの MySQL-Binlog には大量の更新情報が含まれています。そのため、ODS 層では増分データと過去の完全データを使用して、重複排除後に新しい完全データを生成する必要があります。後続の DWD、DWS、ADS も同様の原理です。このプロセスにより大量のデータ再計算が発生し、データ出力の遅延にもつながっています。

• データクエリが遅い原因：Hive 自体に必要なインデックスデータが不足しているため、大規模スループットの計算でも、分単位の遅延を保証したいクエリでも、すべて MR-Job に変換して計算されます。これにより、迅速なデータ探索が必要なシナリオではクエリ結果の出力が遅くなります。

ソリューションの検討

上記の分析から、オフラインデータウェアハウスのデータインクリメンタル更新問題を解決できればリンク計算のパフォーマンスを向上でき、さらにデータテーブルのインデックス機能をサポートできれば、クエリ機能を低下させずにクエリ遅延を削減できます。

• HBase+ORC ベースのソリューション

HBase をデータ更新問題の解決に活用できます。RowKey をプライマリキー、カラムを Column に設定することで、リアルタイムのデータ書き込み機能を提供できます。ただし、HBase の構造上、非プライマリキーカラムのクエリパフォーマンスは非常に低くなります。クエリパフォーマンスを解決するため、HBase テーブルを定期的に HDFS にエクスポートし、特定のフィールドでソートした後に ORC 形式で保存する必要があります（たとえば、時間次テーブルや日次レベルテーブルは日次レベルテーブルとして扱います）。しかし、ORC 形式は単一列の最小値・最大値インデックスのみをサポートしており、クエリパフォーマンスは依然として要件を満たせません。さらに、HBase のデータ書き込みは継続的に発生するため、エクスポートタイミングの制御が困難で、エクスポートプロセス中にもデータが変更される可能性があります。たとえば、12 月 11 日 21:00 以前のデータをデータテーブルの 21:00 パーティションのデータとしてエクスポートする場合、バージョン数、ストレージ容量、フィルタリングによるコンピューティングパフォーマンスなどの要素を考慮する必要があり、システムの複雑性が急激に増加します。同時に、HBase システムの導入により運用コストも増加します。

• データレイク

データレイクは、追加のサービスを導入することなく、主要なコンピューティングエンジン（Flink/Spark など）とデータストレージ（オブジェクトストレージなど）の間で統合できるデータ形式です。リアルタイム Upsert をサポートし、マルチバージョンサポートを提供し、任意のバージョンのデータを読み取れます。

現在、データレイクソリューションには主に DeltaLake、Iceberg、Hudi があります。Alibaba Cloud 上でこれら 3 つのソリューションを調査しました。それぞれの違いと特徴は以下の通りです。

さらに、使いやすさ（DeltaLake はセマンティクスが明確で、Alibaba Cloud がフル機能の SQL 構文サポートを提供し、使いやすい。後者 2 つは使用のハードルが高い）、機能性（Zorder/Dataskipping によるクエリ高速化をサポートするのは DeltaLake のみ）などの側面を考慮し、当社のシナリオを総合的に検討した結果、最終的に DeltaLake をデータレイクソリューションとして選択しました。

4、DeltaLake ベースのオフラインデータウェアハウス

DeltaLake の導入後、オフラインデータウェアハウスのアーキテクチャは以下のようになります。

まず、Binlog は Canal による収集後、自社開発のデータ配信システムを通じて Kafka に書き込まれます。ここで事前に説明が必要なのは、配信システムが Table レベルで Binlog の順序を厳密に維持する必要があることです。理由については後述します。次に Spark を使用してデータをバッチで DeltaLake に書き込みます。最後に、データ取得プラットフォームをアップグレードし、Spark SQL を使用して DeltaLake からデータを取得するようにしました。

DeltaLake の利用プロセスで、以下の重要な技術的課題を解決する必要がありました。

ストリームデータからバッチへの変換

業務シナリオでは、オフラインデータウェアハウスの ETL タスクはデータテーブルのパーティション準備完了をトリガーとして実行されます。たとえば、2021-12-31 のタスクは 2021-12-30 のデータテーブルパーティションの準備完了に依存して実行がトリガーされます。このシナリオは Hive のシステム上では容易にサポートされます。Hive は日付フィールド（dt など）によるパーティショニングを自然にサポートしているためです。しかし、DeltaLake ではデータ書き込みがストリーミングであるため、ストリーミングデータをバッチデータに変換する必要があります。つまり、1 日分のデータが完全に準備できた時点で、対応する日次レベルパーティションの読み取り機能を提供する必要があります。

完全なデータ準備完了の定義方法

ストリーミングデータは一般的に順序が乱れます。順序が乱れた状態では、Watermark メカニズムを採用しても、一定時間範囲内のデータ順序しか保証できません。オフラインデータウェアハウスでは、データは 100% 信頼性が高く、欠落があってはなりません。データソースの順序問題を解決できれば、データ準備完了の解決策ははるかにシンプルになります。日次レベルでパーティショニングする場合、12-31 のデータがあれば、12-30 のデータは準備完了とみなせます。

そこで、当社のアプローチは 2 つのサブ問題に分かれます。

• 順序付けされたストリームデータでバッチデータ境界を定義する

• ストリームデータの順序性を保証するメカニズム

まず前者について、全体の計画は以下の通りです。

• データテーブルの論理パーティションフィールド dt と対応する時間単位情報を設定する。

• Spark がバッチデータを読み取る際、データ内のイベント時間を使用して、上記のテーブルメタデータに従って対応する dt 値を生成する。データストリーム内のイベント時間の値が T+1 に属する場合、データバージョン T のスナップショットがトリガーされ、スナップショットで見つかった対応するデータバージョン情報に基づいてデータが読み取られる。

ストリームデータの順序乱れの解決方法

app-log であれ MySQL-Binlog であれ、ログ自体は順序性があります。たとえば MySQL-Binlog では、単一の物理テーブルの Binlog は必ず順序性があります。しかし、実際の業務シナリオでは、業務システムはデータベース分割とテーブル分割を使用することが一般的です。テーブル分割を使用するシナリオでは、1 つの論理テーブルが Table1、Table2... に分割され、オフラインデータウェアハウスの ODS テーブルでは、業務側の MySQL テーブル分割の詳細とロジックを隠蔽する必要があります。このように、問題はテーブル分割シナリオでのデータ順序性の保証に集約されます。

• 異なるデータベースおよびテーブル、さらには異なるクラスターからの Kafka へのデータ書き込みの順序性を保証する。つまり、DeltaLake への書き込みを行う Spark が論理テーブルの Topic から読み取るデータはパーティション粒度である。

• ODS テーブルの準備完了のタイムリー性を保証する。たとえば、Binlog データがない場合でも、ODS 層のデータはスケジュール通りに準備完了できる。

ここで元のシステムのアップグレードが必要です。プランは以下の通りです。

上記の図に示すように、MySQL クラスターの Binlog は Canal による収集後、特定の Kafka Topic に書き込まれます。ただし、書き込み時に db と Table（テーブル分割 _* サフィックス）に基づいてハッシュでパーティションが決定されるため、単一パーティション内に複数の物理テーブルの Binlog が存在し、DeltaLake への書き込みに非常に不都合です。他のデータアプリケーションとの互換性を考慮し、データ配信サービスを追加しました。

• 論理テーブル名（テーブル分割 _* サフィックス除去）のデータを対応する Topic に書き込み、物理テーブル名でハッシュを行う。単一パーティション内のデータ順序を常に保証し、単一の Topic には 1 つの論理テーブルのデータのみを含める。

• MySQL クラスター内に内部ハートビートテーブルを構築し、Canal 収集の遅延例外を監視する。この機能に基づいて一定のしきい値を設定し、システムに Binlog データがない場合に、システムに問題があるのか、データ自体が存在しないのかを判定する。後者の場合、DeltaLake でセーブポイントをトリガーし、タイムリーにスナップショットをトリガーして、ODS テーブルのタイムリーな準備完了を保証する。



上記のプランを通じて、Binlog ストリームデータを DeltaLake に書き込み、テーブルパーティションの準備完了時間遅延は 10 分未満です。

読み書きパフォーマンスの最適化

次に、DeltaLake の利用プロセスで直面したパフォーマンス問題と対応するソリューションについて説明します。

DPP による書き込みパフォーマンスの向上

DeltaLake は SparkStreamingSQL を通じたデータ書き込みをサポートしています。

レコードのマージと重複排除が必要なため、MERGE INTO による書き込みが必要です。DeltaLake は 2 つのステップでデータを更新します。

• 更新対象ファイルを特定する。デフォルトでは、すべてのファイルと当該バッチの増分データを読み込み、Spark 内で join して更新対象ファイルを関連付ける必要がある。

• マージ後にこれらのファイルを再書き込みし、古いファイルを削除対象としてマークする。

上記の左図に示すように、DeltaLake はデフォルトで前バージョンの全ファイルを読み込むため、書き込みパフォーマンスが非常に低く、Spark の単一バッチでマージ操作を完了できません。

このシナリオに対応するため、DeltaLake をアップグレードしました。DPP を使用したパーティションプルーニングにより、MERGE INTO のパフォーマンスを最適化しました。上記の右図に示す通りです。

• Merge-on 条件を分析し、ソーステーブル内の DeltaLake テーブルのパーティションフィールドに対応するフィールドを取得する。

• 統計によりパーティションフィールドの列挙リストを取得する。

• 前ステップの結果を Filter オブジェクトに変換し、データファイルリストをさらに絞り込むために適用する。

• 最終的なデータファイルリストを読み込み、当該バッチのソースデータと関連付けて、最終的な更新対象ファイルリストを取得する。



DPP 最適化後、Spark バッチ（5 分粒度）の処理遅延は最大 20 分以上から最大 3 分に短縮され、処理時間の長期化による遅延の蓄積問題を完全に解消しました。

Zorder による読み取りパフォーマンスの向上

データ書き込みパフォーマンスの問題を解決した後、データ読み取りパフォーマンスの問題に直面しました。

同じデータ（200 億件以上）を使用し、Hive で計算した平均遅延は 10 分以上でしたが、DeltaLake 使用後の平均遅延は約 11 分でした。分析の結果、主な原因はフィルター列が Zorder でソートされていないことと判明しました。Zorder を有効化すると、遅延は約 24 秒に短縮され、約 25 倍のパフォーマンス向上を実現しました。

Zorder に基づく DeltaLake テーブルのクエリ最適化は、主に 2 つの側面に関わります。

• Dataskipping

• DeltaLake はファイル粒度で各フィールドの最大値/最小値を統計し、データファイルを直接フィルタリングするために使用します。

• Zorder

• データレイアウト手法の一つで、データを再配置し、Zorder フィールドのデータ局所性を最大限に保証します。

Zorder 構築時間の最適化

対象列に対して、Zorder は必要に応じて構築されます。一般的に構築時間は約 30 分です。データがスキューしている場合、Zorder の構築時間は最大約 90 分になります。

Zorder はこれら 2 つの状況に対して最適化されました。

• 一般的に、マルチ列 Zorder では、データセットの複数回走査から 1 回の走査に変更し、構築効率を向上させました。構築時間は平均約 30 分から約 20 分に短縮されました。

• スキューデータの場合、偏った列が存在するバケットを分散化し、構築時間は平均約 90 分から約 30 分に短縮されました。

全体的な効果

半年以上の開発と最適化を経て、DeltaLake ベースのオフラインデータウェアハウスが最近リリースされました。分析のクエリ最適化の向上に重点を置いています。同時に、時間単位のフルデータ要件があるシナリオへのサポートも提供しています。全体的な効果は以下の通りです。

• より高速な準備時間：ODS を DeltaLake に置き換えた後、本番準備完了が午前 2:00〜3:00 から午前 0:10 頃に前倒しされ、2 時間以上前倒しされました。

• より広範な拡張性：ビッグデータは時間単位のフルスコープをサポートする能力を備えました。DeltaLake のインクリメンタル更新機能により、低コストで時間単位のフルスコープ要件を実現でき、従来方式での全データ読み取りの消費を回避できます。現在、一部のコアビジネスに適用して時間単位のフルテーブルを構築しており、同時にタイムリー性は従来の約 40 分から約 10 分に改善されました。

• クエリ速度の向上：アナリストのアドホッククエリの効率向上に重点を置いています。アナリストがよく使用するデータウェアハウステーブルを DeltaLake に移行した後、Zorder を使用してクエリ高速化を実現し、クエリ速度は従来の数分から約 3 分に短縮されました。

5、今後の計画

ジョブヘルプでの DeltaLake の利用において、まだ解決すべき問題が残っています。

• データ修正の効率性を向上させる。

• Hive を使用していた際は、特定の履歴パーティションを個別に簡単に修正できましたが、DeltaLake テーブルの修正では、障害が発生したバージョン以降のすべてのバージョンをロールバックする必要があります。

• Hive エンジンの完全サポート。

• 現在、DeltaLake を使用して、主に過去に Hive のクエリが遅い問題と Presto の複雑クエリ制限の問題を解決し、複雑なクエリと低遅延のソリューションを提供しています。ただし、前述の GSCD や Dataskipping などの機能は Hive でサポートされておらず、ユーザーは Hive と同様に DeltaLake を使用できません。

• Flink アクセスのサポート。

• 当社のストリームコンピューティングエコシステムは主に Flink を中心に構築されています。DeltaLake の導入後、Spark も使用しており、ストリームコンピューティングエコシステムの維持コストが増加しています。