Linkflow builds data lake production practice

1. 背景

Linkflow は、顧客データプラットフォーム (CDP) として、顧客データの収集、分析、実行に至るクローズドループ運用を企業に提供しています。
データ収集エンドポイント (SDK) や、WeChat、Weibo などのサードパーティデータソースを通じて、毎日大量のデータが収集されます。
これらのデータはクレンジング、計算、統合を経て、ストレージに書き込まれます。
ユーザーは柔軟なレポートやタグを通じて永続データを分析・計算でき、その結果は MA (Marketing Automation) システムのデータソースとして、特定のユーザー層に対する精密なマーケティングに活用されます。


Linkflow では、データは不変データ (Immutable Data) と可変データ (Mutable Data) に分類され、これらは分析処理に参加します。
関連するテーブルは約 12 あり、不変データのデータ量は非常に大きく、数十億レコードに達する可能性があります。
従来のビッグデータシステムでは、不変データはファクトデータ、可変データはディメンションデータに相当します。
しかし実際のビジネス運用では、ユーザーの属性や注文の金額とステータスなどはすべて更新可能であり、これらのデータ量も非常に膨大になることが多く、Linkflow では数十億レコードに達します。
可変データはこれまで MySQL を通じて管理されてきました。
データ管理が容易であり、ビジネス連携も簡単だからです。


しかし、問題も明らかです。


データの断片化。
MySQL の大規模テーブルに対するオンライン DDL 操作はリスクが高いため、ビジネスの複雑化に伴い、新しいサブテーブルを追加してビジネス属性を拡張する必要が生じることがよくあります。
つまり、完全なユーザーデータが複数のテーブルに散在することになります。
クエリには非常に不利な状況です。

多次元クエリを実現できません。
リレーショナルデータベースは多次元クエリを得意とせず、すべてのフィールドにインデックスを張ることも現実的ではないため、OLAP クエリエンジンをサポートするデータコンポーネントが必要です。
将来の独立した拡張性も考慮し、コンピューティングとストレージの分離アーキテクチャを優先的に採用しています。


2. CDC とデータレイク

CDC (Change Data Capture) は、変更データを識別・追跡するためのソフトウェアデザインパターンであり、変更されたデータに対してアクションを実行できるようにします。
実際、2 年前には canal を使用して MySQL データを異種ストレージに冗長化する経験をしていましたが、ビッグデータストレージと統合できることには気づいていませんでした。
canal の使用中にいくつかのパフォーマンス問題が発見され、オープンソースコミュニティもほぼメンテナンスされていない状態だったため、新アーキテクチャの立ち上げ前に Maxwell と Debezium を調査しました。
その過程で、Flink の親会社である Ververica のオープンソースプロジェクト flink-cdc-connectors [1] に注目しました。
このプロジェクトは、Debezium を binlog 同期エンジンとして Flink タスクに埋め込み、フロータスク内で binlog メッセージのフィルタリング、検証、データ統合、フォーマット変換を簡単に実現でき、優れたパフォーマンスを発揮します。
将来的に行動データとの双方向ストリーム結合や、CEP を通じたシンプルなリスク管理も可能であることを考慮し、Flink 上の Debezium CDC ソリューションを最終的に選択しました。


MySQL には多数のデータトピックが存在するため、フロータスク内でデータルーティングも実装しています。
異なるトピックの変更データは対応する Kafka トピックにルーティングされ、Kafka を ODS として使用します。
このアプローチには多くの利点があります。
第一に、可変データに対する各変更のプロセスを明確に観察できることです。
第二に、データの再生が可能なことです。
連続的な変更を重ね合わせた結果が最終状態となります。


次に考慮すべきはデータの保存先です。
前述の「コンピューティングとストレージの分離」の原則と合わせて、これはデータレイクが提供する利点でもあります。
データレイクは一般的にファイルシステムベースのストレージ (オブジェクトストレージや従来の HDFS) を使用して構築されます。
複数のデータレイクソリューションを比較検討した結果、Apache Hudi を選択しました。
理由は以下の通りです。


Hudi は HDFS における upsert のソリューションを提供し、リレーショナルデータベースと同様の操作感を実現しています。
更新可能なデータとの相性が良く、MySQL の binlog セマンティクスにも適合しています。


増分クエリにより、過去 30 分間や 1 日以内に変更されたデータを簡単に取得できます。
これはオフラインコンピューティングタスクにとって非常に有利です。
全量データではなく変更されたデータのみを処理すればよいため、マシンリソースと時間を大幅に節約できます。


メタデータは Hive にリアルタイムで同期され、「データレイクへの取り込みと同時に検索可能」な環境を実現します。


COW と MOR という 2 つの異なるユースケースにそれぞれ最適化されています。


Hudi コミュニティはオープンで、迅速にイテレーションされています。
インキュベーション期間中に AWS EMR に統合され、その後 Alibaba Cloud DLA データレイク分析 [2]、Alibaba Cloud EMR [3]、Tencent Cloud EMR [4] にも統合されました。
将来性が期待でき、国内の技術交流グループでの Apache Hudi に関する議論も非常に活発で、Hudi ベースのデータレイクを構築する国内企業が増えています。


Hudi を統合した後、アーキテクチャは以下のように進化しました。


データテーブルには COW (コピーオンライト) モードを選択しました。
読み取りが多く書き込みが少ない特性を考慮し、クエリプロセスを可能な限り高速化する必要があるためです。
MOR (マージオンリード) 戦略はクエリ側でのパフォーマンスがまだやや劣る上、データレイテンシにサブ秒要件がないため、最終的に COW を選択しました。


最上位レイヤーでは Presto を分析エンジンとして使用し、データのアドホッククエリ機能を提供しています。
使用している Hudi バージョンは 0.6.0 で、Flink との統合がまだリリースされていなかったため、Flink + Spark のデュアルエンジン戦略を採用し、Spark Streaming で Kafka のデータを Hudi に書き込む必要がありました。


3. 技術的な課題

PoC を実施して前述のアーキテクチャ設計を確定しましたが、実際の実装プロセスでは多くの課題に直面しました。


3.1 CDC 操作モードのカスタマイズ

■ フルモード

Debezium の大きな利点の一つは「バッチとストリームの統合」です。
スナップショット段階では、テーブル全体をスキャンして binlog 増分ログと一致するメッセージにデータを再生するため、同じコードでフルデータと増分データの両方を同時に処理できます。
しかし実際のビジネス運用では、履歴テーブルの数やデータ量が多い場合、スナップショット段階が非常に長時間続きます。
プロセスが予期せず中断されると、次回に最初のテーブルから再スキャンする必要があります。
完全なスナップショットに数日かかる場合、この規模のリトライは受け入れられないため、レジューム機能のようなメカニズムが必要です。
Debezium の公式ドキュメントを調査した結果、snapshot.include.collection.list パラメータを見つけました。


「table.include.list で指定されたスキーマ名のうち、スナップショットを取得したいものに一致する正規表現の、省略可能なカンマ区切りリスト」

したがって、スナップショットが中断された後、残りのスキャン対象テーブルをこのパラメータで渡すことで「リレー」機能を実現できます。
ただし注意点として、スナップショット段階を何回リトライしても、増分 binlog の位置は必ず最初のスナップショットの位置でなければなりません。
そうでないとデータが失われます。
これによりもう一つの問題が生じます。
中断とリレーを繰り返してスナップショットが完了すると、Debezium はその (最初のものではない) スナップショットの binlog 位置から自動的に増分同期を開始してしまいます。
これは求める結果ではなく、スナップショット終了後にタスクを直接終了させる必要があります。
Debezium のドキュメントを詳しく探しましたが、そのような機能は見つかりませんでした。
しかしソースコードを参照しているうちに、実現方法があることがわかりました。


■ 増分モード
スナップショット終了後にタスクが自動停止する場合、増分同期を続けるにはタスクを手動で再起動する必要があります。
同時に、増分モードでは MySQL の binlog ファイルと具体的な位置を指定できる必要があります。


3.2 パッチ更新 (Patch Update)

ここで、上書き更新と部分更新の違いを説明する必要があります。
これは RESTful セマンティクスに対応しています。
PUT は上書き更新で、呼び出し元が完全なリソースオブジェクトを提供する必要があります。
理論上、PUT を使用する場合、完全なリソースオブジェクトが提供されなければ、欠落フィールドはクリアされるべきです。
PATCH は部分更新に対応し、呼び出し元は更新が必要なフィールドのみを提供し、完全なリソースオブジェクトは不要です。
帯域幅を節約できる利点があります。


Hudi ではデフォルトで上書き更新のみがサポートされていますが、実際のビジネスでは、収集エンドポイントから報告されるデータに完全なビジネスオブジェクトを含めることができない場合があります。
たとえばユーザーの年齢増加のようなデータでは、報告には 1 つのフィールドの情報しか含まれません。


この場合、まず rowkey=123 のデータ内容を取得し、更新対象の内容とマージしてから書き込む必要があります。
マージ時、書き込みデータのフィールドが空でなければマージを実行します。
Hudi はデフォルトで OverwriteWithLatestAvroPayload の combineAndGetUpdateValue メソッドを使用します。


「最新のデルタレコードでストレージを上書きするだけ」

前方互換性のため、データ開発担当の Karl が OverwriteNonDefaultsWithLatestAvroPayload クラスを追加し、combineAndGetUpdateValue をオーバーライドしてこの問題を処理しており、コミュニティにもフィードバック済みです [HUDI-1255] Add new Payload (OverwriteNonDefaultsWithLatestAvroPayload) for updating specified fields in storage [5]。
実際、コミュニティには同様のニーズが多数存在しており、たとえば [HUDI-1160] Support update partial fields for CoW table [6] などがあり、より多くの開発者によってこの機能がさらに完全なものにされることが期待されています。


ただし制限もあります。
フィールドを実際に null 値に更新したい場合、OverwriteNonDefaultsWithLatestAvroPayload では実現できません。


同時に、時間ベースのコンパクションスケジューリング戦略を追加することで、コミュニティのコンパクション戦略も補完しました。
増分提出の数だけでなく、時間に基づいてコンパクションを実行できるようになり、指定時間内のコンパクションにより柔軟性がもたらされています。
この取り組みもコミュニティにフィードバック済みです [HUDI-1381] Schedule compaction based on time elapsed [7]。


3.3 同じ行キーデータのバッチ内マージ

CDC の特徴としてリアルタイムでデータ変更を監視する機能があるため、たとえば注文のステータスが数分以内に複数回変更される可能性があります。
Spark Streaming のマイクロバッチ処理の特性と相まって、時間ウィンドウ内で同じ行キーを持つ大量のデータが高確率で取得されます。
そのため、Streaming タスク内で同じ行キーのデータをバッチマージしています。
全体的なロジックは、Hudi が Bloom を使用して行キーの存在を判定するロジックと似ています。
特に注意すべきはタイミングの問題です。
データの重ね合わせは必ずタイムスタンプの順序に厳密に従わなければならず、そうでなければ古いバージョンのデータが新しいバージョンを上書きしてしまいます。


3.4 スキーマ進化

ビジネス開発と柔軟性の要件から、テーブルフィールドの拡張 (スキーマ進化) は必須です。
Hudi はこの点を考慮しており、Hudi の wiki [8] から以下の情報を得ました。


スキーマ進化は大まかに 4 つのタイプに分けられます。


下位互換 (Backwards compatible)：古いデータが新しいスキーマで読み取り可能で、フィールドに値がない場合はデフォルト値が使用されます。
Hudi が提供する互換性方式でもあります。


上位互換 (Forwards compatible)：新しいデータが古いスキーマで読み取り可能で、Avro が新しく追加されたフィールドを無視します。
上位互換性を保つには、削除されるフィールドに必ずデフォルト値が必要です。


完全互換 (Full compatible)：上位互換と下位互換の両方をサポートします。
完全互換性を保つには、追加するフィールドはデフォルト値を持つものに限られ、削除するフィールドもデフォルト値を持つものに限られます。


互換性チェックなし (No Compatibility Checking)：通常、フィールドの型を強制的に変更する必要がある場合です。
全量データ移行が必要になるため、推奨されません。


本番環境では、スキーマを変更することでフィールド拡張要件を実現できます。
ただし後で問題も発見されます。
フィールドが多すぎると単一ファイルが非常に大きくなり (128 MB を超える)、書き込みが非常に遅くなります。
極端な場合、1,000 列を超えるファイルの書き込みには数時間かかります。
将来的には、フィールドのリサイクルや垂直テーブル分割などの最適化ソリューションを検討し、単一ファイルのフィールド数を削減していく予定です。


3.5 同時クエリと書き込みによる例外

これはクエリ側の問題です。
Presto で Hive テーブルをクエリする際、Hudi のメタデータファイルが見つからない例外が発生し、Hudi 内部で NPE が発生します。


前述の情報に基づき、クエリ中にメタデータ情報が変更されている疑いがありました。
コミュニティに相談した後、HoodieROTablePathFilter 内の hoodiePathCache をスレッドセーフな ConcurrentHashMap に変更し、再パッケージして hudi-hadoop-mr.jar と hudi-common.jar を取得し、presto/plugin/hive-hadoop2 ディレクトリに配置して Presto を再起動しました。
その後 NPE は発生しなくなりました。


4. 効果

アーキテクチャ設計時に掲げたデータレイクのビジョンを振り返ってみましょう。


可変データのサポート。

スキーマ進化のサポート。

コンピューティングとストレージの分離、複数クエリエンジンのサポート。

増分ビューとタイムトラベルのサポート。

これらの機能は Hudi で基本的に実現されています。
新アーキテクチャの完成後、以前のシステムと比べてデータ遅延とオフライン処理性能が大幅に改善されました。
以下の通りです。


リアルタイムデータ書き込みプロセスが簡素化されました。
以前は更新操作が煩雑でしたが、現在は開発時に追加や更新の操作を基本的に意識する必要がなく、開発者の精神的負担が大きく軽減されています。

リアルタイムデータがデータレイクに取り込まれてからクエリ可能になるまでの時間が短縮されました。
COW テーブルモードを使用していますが、実際のテストではデータレイクへの取り込みからクエリまでの適時性は高く、基本的に分レベルでした。

オフライン処理性能が向上しました。
Hudi の増分ビュー機能に基づき、日次オフラインタスクで過去 24 時間に変更されたデータを簡単に取得できるようになり、処理対象データ量が削減されたことで処理時間が短縮されています。


5. 将来の計画

5.1 Flink との統合

前述の「やむを得ない」デュアルエンジン戦略は非常に悩ましく、運用・開発方法を統一できないため、Hudi の公式 Flink 統合の進捗を注視しています。
最近、RFC-24: Hoodie Flink Writer Proposal [10] が公開され、Hudi バージョン 0.8.0 で Flink の機能が深く統合されました。
今後の統合バージョンで性能が大幅に向上し、処理エンジンを Flink に統一してデュアルエンジンモードから脱却できることを期待しています。


5.2 同時書き込み

メタデータの一貫性を保証するため、バージョン 0.8.0 より前の Hudi ファイルは同時書き込みをサポートしていません。
しかし実際のアプリケーションでは、データレイク内の多くのデータがリアルタイムデータだけでなく、オフライン計算で取得する必要がある場合もあります。
テーブル内の一部のフィールドが CDC によって直接反映され、他の部分がオフラインタスクの計算結果である場合、同時書き込みのニーズが生じます。


現在、2 つの方法で回避しています。


垂直テーブル分割。
2 つのファイルに分離し、CDC データを Spark Streaming で書き込み、オフライン計算結果を別のファイルに書き込むことで同時書き込みを回避します。


CDC メッセージへのシミュレーション。
クエリパフォーマンスのためにテーブルを分割できない場合、オフライン計算結果を CDC メッセージにシミュレートして Kafka に書き込み直し、Spark Streaming で Hudi に書き込みます。
ただし、オフラインタスクの結果が最終ストレージに反映されるまでに時間がかかるという欠点があります。


最近リリースされた Hudi バージョン 0.8.0 では、楽観的ロックに基づく同時書き込みモードが既にサポートされています。
ファイルレベルの競合検出により同時書き込み要件を十分に満たせるため、今後テストして効果を確認する予定です。


5.3 パフォーマンス最適化
前述の大きなファイルや頻繁な GC などの問題について、全体的に書き込みのボトルネックは主に 2 か所で発生していることがわかりました。


■ インデックス
現在 HoodieGlobalBloomIndex を使用していますが、インデックスの構築とクエリに時間がかかります。
公式では 3 つのインデックス実装が提供されています。

■ 更新

大きなファイルの問題に加え、アップサートの遅さは CDC の特性にも関係しています。可変データの更新範囲は実際には予測不可能で、極端な場合、更新対象の 1,000 件のデータが 1,000 の異なるファイルに属していると、コード最適化だけでは性能改善が難しく、CPU リソースを増やして処理の並列性を向上させるしかありません。以下の観点から取り組む予定です。

パラメーター調整により、ファイル数とサイズのバランスを取る方法があるかどうか。
一部のビジネステーブルで MOR モードの使用を試みる。MOR は更新時にまずログファイルに書き込まれてから Parquet にマージされるため、理論上は Parquet ファイルの上書き頻度を減らせます。

ビジネス側のトレードオフを議論し、より高速な書き込みとのバランスを取ります。