Programming Delta Lake real-time data warehouse application practice

Walnut Programming の紹介

Walnut Coding は 2017 年 8 月 9 日に設立されました。子ども向けプログラミング教育業界のリーダーとして、「すべての子どもに学ぶことを愛し、学び方を知ってもらい、質の高い教育を手が届くものにする」というミッションを常に堅持し、技術的手段を通じて子どもたちの教育を推進することに注力しています。AI を活用したマンツーマン dual-teacher 教育モデルと 10 段階の高度なカリキュラム体系により、適性に合わせた大規模な教育を実現し、「中国の子どもたちの学習能力を触発する」ことを目指しています。2019 年 8 月時点で、Walnut Programming は有料生徒数で最大規模の子ども向けプログラミング教育機関となり、65 万人以上の子どもたちが学習意欲を高め、プログラミングスキルを磨き、優れた思考習慣を育むことを支援しています。生徒の再購入率は 91% を超え、授業完了率は 98% に達し、1,873 万点のオリジナル作品がオンラインで公開されています。

1. ビジネスの現状
ビジネスニーズ
授業は決まった時間に開始されます。授業時間中、担任教師は生徒の学習状況をリアルタイムまたは準リアルタイムで把握する必要があります。
データ集計のディメンションは通常、クラス単位および学期単位で集約され、時間範囲は数か月から 1 年に及ぶことがあります。
ビジネスの変化は急速であり、ビジネス変更に伴う指標のロジック変更にも迅速に対応する必要があります。
データソース

構造改革前の仕組み
既存の指標は、Kafka / MySQL のデータを HDFS に書き込み、Hive オフラインバッチ処理を使用して 10 分ごとに実行し、過去の累積指標を周期的に計算してから、データを定期的に MySQL に同期し、バックグラウンドクエリに提供しています。以下の図の通りです。

直面している課題
計算対象のデータ量が増加するにつれ、ビジネスが求める更新頻度の要件を徐々に満たせなくなっています。

Apache Sqoop を使用した全量データ同期は、ビジネスの MySQL データベース / HDFS に負荷をかけます。
Apache Sqoop を使用した増分同期では、通常、特定の時間フィールド（更新時刻など）を使用して新たに変更されたデータを同期することしかできません。そのため、パーティションテーブルを作成する際に、より複雑なオフラインマージが必要になります。
データが大きくなるにつれ、同期と処理の時間が長くなり、ビジネスのリアルタイム要件を満たせなくなります。
2. リアルタイムデータウェアハウスソリューションの検討
オフライン同期ソリューションではビジネスニーズを満たせなくなったため、リアルタイムソリューションへの移行が計画され、いくつかの検討が行われました。

ストリームコンピューティング移行の課題
開発サイクルが長い
既存のオフラインタスクは、多くは数百行に及ぶ SQL でロジックが複雑です。すべてのロジックをストリームコンピューティングに移行するのは、開発が比較的困難で、変換コストも高くなります。
たとえば、オフライン増分同期では、まずベースデータの全量を同期する必要があります。

その後、増分のバイナリログデータを消費し、Hive 外部テーブルにストリーム書き込みしてから、2 つのテーブルをマージします。

Delta Lake を使用すれば、ストリーミング SQL 1 つでリアルタイム増分同期を実現できます。

データリカバリが困難
オフラインタスクの場合、データリカバリはタスクを再実行するだけで済みます。

しかし、ストリームコンピューティングでは、データが異常またはロジックが変更され、全量データを再実行する必要がある場合、既存データをオフラインで補完してからリアルタイムデータとユニオンするしかありません。Kafka はすべての既存データを保存できず、最初から消費してデータを追跡するには長時間を要するためです。

迅速なリカバリのニーズに対応するため、すべての指標について、リアルタイム用とオフライン用の 2 セットのコードを最初から用意しておく必要があります。Lambda アーキテクチャに似ています。

データ検証が困難
Kafka は通常、ビッグデータアーキテクチャ内でメッセージキューとして機能し、データ保持期間は短くなっています。既存データの全量は Kafka を消費して HDFS に書き込まれます。ある指標を 1 か月間計算した後に計算結果に異常が見つかった場合、その時点の Kafka データに問題があったのか、計算ロジックに問題があったのかを追跡するのは困難です。HDFS データを使用して調査することもできますが、HDFS 内のデータがその時点の Kafka データと一致している保証はありません。

求められる機能
ストリームコンピューティングへの移行にはいくつかのコストと課題があるため、リアルタイムコンピューティングソリューションに対して以下の機能要件が提示されました。

柔軟な開発
インターネット企業のビジネス展開は速く、人的リソースは比較的に逼迫しています。低いコストと短い期間で新しい指標を開発し、ビジネスの俊敏性に対応する必要があります。

既存データの再実行が容易
ビジネス指標の定義は頻繁に変更されます。変更後、または新しいデータ指標は、最初から消費する必要があります。しかし、通常、大量の既存データがあり、一般的なリアルタイムデータソースである Kafka はすべての既存データを保存できません。

データ異常時のトラブルシューティングが容易
オフラインデータウェアハウスを例に取ると、数百行の SQL を分割して実行し、段階的に確認できます。Flink はメトリクスを記録することで中間プロセスを取得できます。

3. Delta Lake ベースのリアルタイムデータウェアハウスソリューション
Delta Lake
Delta Lake は、米国の Databricks が提供するオープンソースのデータレイク技術です。Apache Parquet を基盤に、メタデータ管理 / トランザクション / データ更新 / データバージョンバックトラッキングなどのデータ管理機能を充実させています。Delta Lake を使用することで、ストリーム処理とバッチ処理をシームレスに連携させ、準リアルタイムデータパイプラインを迅速に構築できます。
image.png

現在、Alibaba Cloud E-MapReduce (EMR) チームは、Delta Lake の機能とパフォーマンスを多数最適化し、Spark と深く統合しています。主に以下の側面です。詳細は EMR 公式ドキュメントをご参照ください。

SparkSQL で Update / Delete / Merge Into / Optimize / Vacuum などの構文をサポートし、Delta Lake を操作
独自開発の SparkStreaming SQL で、Delta Lake 関連の DML 操作をサポート
Hive & Presto On Delta Lake
Delta Lake On OSS (Alibaba Cloud Object Storage Service)
Delta Lake トランザクション競合検出の最適化
DataSkipping & Zorder パフォーマンス最適化
image.png
SparkStreaming SQL
Alibaba Cloud EMR チームは、StructStreaming をベースに SparkStreaming SQL を開発しました。ユーザーは SQL を使ってストリームジョブのロジックを簡単に記述でき、開発の敷居を大幅に下げています。詳細は SparkStreaming SQL 公式ドキュメントをご参照ください。

バッチ統合エンジン
SparkSQL / SparkCore の最適化を再利用可能
豊富な SQL サポート

豊富な UDF サポート
Hive UDF / ウィンドウ関数
豊富なデータソースサポート

Delta Lake との深い統合
Delta Lake の利用シーンに合わせて、いくつかの新機能をサポート（ストリーム動的パーティションテーブル書き込みなど）
リアルタイムデータウェアハウスソリューション
アーキテクチャ設計
Delta Lake + SparkStreaming SQL をベースに、リアルタイムデータウェアハウスのパイプラインを迅速に構築できます。以下の通りです。

ODS レイヤー
ODS データは主にリアルタイムトラッキングポイントデータ、CDC のバイナリログなどです。
DIM ディメンションテーブル
DW レイヤー
DW レイヤーは主に軽度集約データで、ユーザーディメンションのコースや課題などです。

主に再利用されるのは DW レイヤーのデータであるため、各指標について、集約するかどうか、どのディメンションに集約するか、ディメンションテーブルを関連付けるかどうかを総合的に考慮する必要があります。
DW レイヤーは 2 つのタイプに分かれます。

a. ビジネスが単純で、基本的に変更がない場合。Kafka に直接書き込みます。
b. ビジネスロジックが複雑で、データが変更される場合、Delta Lake に書き込みます。実際には、Kafka への直接書き込みが最もシンプルなソリューションですが、柔軟性は非常に低く、既存データの追跡や修正ができません。DW レイヤーに Delta Lake を導入することで、ストリームバッチ統一データソースや既存パーティションデータのデータリカバリなどの機能を実現できます。

DM レイヤー

DM レイヤーは最終的なレポート表示指標です。DW レイヤーの Delta テーブルをデータソースとして、再度集約した後、シンクをデータベースに送信して表示します。
備考：
EMR チームはストリーミング Merge Into 機能を提供しており、CDC が SparkStreaming SQL を記述することでバイナリログを Delta テーブルにリプレイできます。
詳細は CDC 同期ドキュメントをご参照ください。

課題の最適化
Delta Lake の使用過程で、いくつかの課題も発見されました。詳細なソリューションと推奨は以下の通りです。

小ファイルの問題
CDC ストリーミング Merge によるバイナリログのリプレイ過程で、小ファイルが継続的に生成されるため、一部の小ファイルを処理する必要があります。EMR はいくつかの最適化ソリューションを提供しています。

シリアル自動コンパクション機能を追加
CDC ストリーミングジョブの実行中に、一定のポリシーに従って小ファイルをマージ・コンパクション
アダプティブ実行を使用
アダプティブ実行スイッチを有効にすることで、Merge 処理で生成される小ファイルを効果的に削減できます。たとえば、1 バッチあたり 100 個の小ファイルを 1~2 個に削減します。
コンパクション競合の問題
シリアル自動コンパクション機能を使用しない場合、Delta テーブル上で定期的に手動コンパクションを実行して小ファイルをマージする必要がありますが、トランザクション送信時にコンパクションと CDC ストリームジョブのトランザクション送信の間で競合が発生することが多く、CDC ストリームまたはコンパクションが失敗する場合もあります。これについてもいくつかの最適化と推奨があります。

Delta カーネルの競合メカニズムを最適化し、CDC フローが安定して実行され、コンパクションにより停止しないようにする
パーティションテーブルを使用してパーティション単位でバッチコンパクションを実行し、競合の確率を低減
データベーステーブルの更新 / 削除操作が少ない時間帯にコンパクションを実行（EMR ワークフローのスケジューリングを利用可能）
EMR ワークフローのジョブリトライ機能を使用して、コンパクションのトランザクション送信失敗時にリトライ
アーキテクチャ設計の補足説明
• なぜ ODS から直接計算しないのか
Walnut Programming の出席率指標を例に取ります。データソースは Kafka のトラッキング Topic です。計算対象の指標には、個人ディメンションからクラスデータ、学期ディメンション、クラスディメンション、学期ディメンション、マーケティングチャネルディメンションが含まれます。
各ディメンションはすべてのトラッキングポイントデータを消費し、そこからクラス関連のイベントを抽出する必要があります。さらに、各ディメンションの計算プログラムは、学期、クラス、単位などの HBase / MySQL 関連ディメンションテーブルをクエリする必要があります。
全体のロジックが調整された場合、たとえばテストクラスデータをフィルタリングする際、ODS レイヤーからデータをフィルタリングすることはできません（そのようにするとデータが最下層から失われ、後で追跡できなくなるため）。すべてのプログラムを再調整して、このフィルタリングロジックを追加する必要があります。

• データリカバリの仕組み
理想的には、リアルタイムとオフラインで同じ SQL、同じ計算ロジック、同じデータソースを使用し、オフラインスクリプトでいつでも既存データを再実行できるようにすることです。しかし、どのフレームワークもこれを完全にサポートしているわけではありません。いわゆるストリームバッチ統合はエンジンレベルでの統合です。たとえば、Spark のストリーミングと SQL はいずれもバッチ方式であり、ストリームはより小さなバッチに過ぎません。一方 Flink は、バッチデータをストリーム方式で処理することを望んでおり、バッチは境界のあるストリームに過ぎません。SQL API レベルでは、ストリームとバッチの差異は非常に大きいです。Delta Lake のパーティションテーブルに基づき、DW レイヤーのリアルタイムデータを時間単位でパーティショニングすることで、オフラインジョブがいつでも過去のパーティションデータをリストアできます。ただし、DW レイヤーで収集されるデータ量は比較的少ないため、リカバリ後にストリームジョブで最初から再消費することも可能です。

4. ビジネス効果
Walnut Programming の一部データウェアハウスの本番環境に Delta Lake リアルタイムデータウェアハウスを導入した後、新しいアーキテクチャに基づいていくつかのビジネス統計指標が生成され、指標の更新遅延は数十分から 1 分以内に短縮されました。担任教師は生徒の学習状況をより迅速に把握し、学習進捗をタイムリーにフォローできるため、教育品質が大幅に向上しました。
CDC の適用後、データ同期遅延は 30 分から 30 秒に短縮され、同時に Apache Sqoop の高同時実行同期によるビジネスデータベースへの影響も解消されました。データアナリストは Ad-Hoc クエリ時にリアルタイムのビジネスデータを取得できるようになり、データ分析の効果が大幅に向上し、ビジネス展開をよりタイムリーに指導できます。

5. 今後の計画
現在のビジネスアプリケーション効果に基づき、ビッグデータチームは今後もビジネススコープ内のすべてのリアルタイム指標を整理し、リアルタイムデータウェアハウスの各レイヤーの構造をさらに最適化し、Delta Lake ベースのリアルタイムデータウェアハウス構築の包括的な応用を推進します。
Delta Lake のモード実行とタイムトラベルの特性に基づき、機械学習シナリオでの Delta の応用をさらに推進し、より信頼性が高く、拡張が容易なデータパイプラインを構築します。