Practice of Accelerating Hive Daily Table Production Based on Flink

1. プロジェクトの背景

SmartNews は機械学習を活用したインターネット企業です。2012 年に東京で設立され、米国および中国にもオフィスを構えています。8 年以上の開発を経て、SmartNews は日本における No.1 ニュースアプリ、米国における最も急成長しているニュースアプリへと成長し、世界 150 以上の国と市場をカバーしています。2019 年初頭の統計によると、SmartNews の iOS 版および Android 版は世界で累計 5,000 万回以上ダウンロードされています。

過去 9 年間にわたり、SmartNews は Airflow、Hive、EMR などの技術スタックを基盤に、大量のデータセットを構築してきました。データ量の増加に伴い、これらのオフラインテーブルの処理時間が徐々に長期化しています。さらに、ビジネス側のイテレーションの加速により、テーブルのリアルタイム性能に対してより高い要件が課されるようになりました。そこで、SmartNews は社内で Speedy Batch プロジェクトを立ち上げ、既存のオフラインテーブルの生成効率の高速化に取り組みました。

本稿は Speedy Batch プロジェクトの一例として、ユーザーのアクションテーブルを高速化した実践例を紹介します。

アプリから送信されるユーザーアクションログは、Hive ジョブによって毎日、日次テーブルとして生成されます。このテーブルは他の多くのテーブルのソースであり、非常に重要です。このジョブの実行には 3 時間を要し、多くの上流テーブルのレイテンシを増加させ、データサイエンティストやプロダクトマネージャーなどのユーザー体験に大きな影響を与えています。そのため、これらのジョブを高速化し、テーブルをより早く利用できるようにする必要があります。

同社のビジネスは基本的にパブリッククラウド上で運用されており、サーバーの元ログはファイル形式で日単位にパーティション分割されたクラウドストレージにアップロードされます。現在のジョブは Airflow によって EMR 上でスケジュール実行され、Hive の日次テーブルが生成されてクラウドストレージに保存されます。

2. 問題の定義

1. 入力
ニュースサーバーは 30 秒ごとに元ログファイルをアップロードし、ファイルは対応する日付および時間のクラウドストレージディレクトリに格納されます。

2. 出力
元ログは ETL 処理を経て、日 (dt) と action の 2 レベルのパーティションに分割されて出力されます。アクションは約 300 種類あり、固定ではなく頻繁に増減します。

3. ユーザー
このテーブルの利用は広範囲かつ多岐にわたります。Hive からのクエリ、Presto、Jupyter、Spark からのクエリがあり、これらがすべてのアクセス方法であるとは限りません。

3. プロジェクトの目標

actions テーブルのレイテンシを 3 時間から 30 分に短縮する
下流ユーザーに対して透過的であること。透過性は以下の 2 つの側面に分けられます。

機能面：ユーザーはコードを変更する必要がなく、完全に透過的です
パフォーマンス面：新しいプロジェクトで生成されたテーブルが、下流の読み取り時にパフォーマンスを劣化させないこと

4. 技術選定

本プロジェクト以前にも、このジョブに対して複数の改善が行われましたが、目立った効果は得られませんでした。

試みたソリューションには、リソースを増やしてマシンを追加する方法がありましたが、クラウドストレージの IOPS 制限に直面しました。各プレフィックスは最大 3,000 の同時読み書きまでしかサポートしません。この問題は出力段階で特に顕著で、複数の reducer が同時に同じ action サブディレクトリに出力する際に制限に遭遇しやすくなります。また、時間単位で前処理を行ってから毎日の早朝に日次テーブルへマージする方法も試しましたが、マージプロセス自体にも多くの時間がかかり、全体的なレイテンシは依然として約 2.5 時間で、十分な効果は得られませんでした。

サーバー側のログがニアリアルタイムでクラウドストレージにアップロードされていることを踏まえ、チームはストリーム処理のアイデアを提案しました。1 日分のデータを待つバッチジョブの模式を放棄し、1 日を通して計算を分散させることで、日終了後の処理時間を短縮するというものです。チームは Flink に関する豊富な経験があり、Flink は最近 Hive に対して多くの改善が行われているため、Flink ベースのソリューションを採用することにしました。

5. 技術的な課題

課題は多岐にわたります。

1. RCFile 形式の出力

現在の Hive テーブルのファイル形式は RCFile です。ユーザーへの透過性を確保するため、既存の Hive テーブルに対してインプレースアップグレードを行うしかなく、現在のテーブルを再利用する必要があるため、Flink の出力も RCFile 形式に準拠しなければなりません。Hive テーブルは 1 つの形式しか持てないからです。

RCFile はバルク形式（行形式に対応）に属し、各チェックポイントごとに 1 回出力する必要があります。5 分ごとにチェックポイントを設定した場合、各アクションは 5 分ごとにファイルを 1 つ出力することになり、結果ファイルの数が大幅に増加し、下流の読み取りパフォーマンスに影響を与えます。特に低頻度のアクションでは、ファイル数が数百倍に増加します。Flink のファイルマージ機能についても調査しましたが、それは単一チェックポイント内の複数シンクデータの統合であり、我々の課題を解決するものではありません。必要なのはチェックポイント間のファイルマージです。

チームは行形式（CSV など）で出力し、カスタム Hive SerDe を実装して RCFile と CSV の互換性を持たせる案も検討しました。しかし、その場合、各クエリシナリオごとに Hybrid SerDe を実装する必要があります。たとえば、Presto 用、Spark 用などです。このアイデアはすぐに断念しました。

一方で、それほど多くのリソースを投入できません
他方で、その種のソリューションはユーザーにも感知されます。結局、ユーザーがカスタム SerDe をインストールする必要があるからです
以前、新しい形式のテーブルを生成する案も提案しましたが、ユーザーへの透過性が十分でないため却下されました

2. パーティションの検知性と整合性

下流ジョブがその日のパーティションの準備完了をどのように検知するか。actions テーブルは dt と action の 2 レベルパーティションに分かれています。action は Hive の動的パーティションに属し、数が多く固定されていません。現在の Airflow の下流ジョブは insert_actions Hive タスクの完了を待ってから実行を開始します。これは問題ありません。insert_actions が終了すれば、すべての action パーティションが準備完了だからです。しかし、Flink ジョブには終了シグナルがなく、dt=2021-05-29/action=refresh のようにパーティションを 1 つずつ Hive に送信するしかありません。アクションの数が多いため、パーティションの送信プロセスは数分間続く可能性があります。したがって、Airflow ジョブに dt レベルでパーティションを検知させることはできません。一部のアクションしか準備できていない段階で下流がトリガーされる可能性があるからです。

3. クラウドストレージファイルのストリーミング読み取り
プロジェクトの入力は継続的にアップロードされるクラウドストレージファイルであり、MQ (メッセージキュー) からのものではありません。Flink は FileStreamingSource をサポートしており、ストリーミング方式でファイルを読み込めますが、これは定期的にディレクトリをリストして新しいファイルを検出する仕組みです。しかし、このソリューションは我々のシナリオには適していません。ディレクトリが非常に大きく、クラウドストレージのリスト操作が正常に完了しないからです。

4. Exactly Once の保証
actions テーブルの重要性を考慮すると、ユーザーはデータの損失や重複を受け入れられないため、ソリューション全体で Exactly Once の処理を実現する必要があります。

6. 全体の計画と課題への対応

1. RCFile の出力と小ファイルの回避
最終的に選択したソリューションは 2 段階で構成されます。最初の Flink ジョブは JSON (行形式) で出力し、その後別の Flink ジョブで JSON を RC 形式に変換します。これにより、Flink が適切なサイズの RC ファイルを直接出力できない問題を解決します。

JSON の中間出力により、ローリングポリシーで出力ファイルサイズを制御できます。複数のチェックポイントにわたってデータを蓄積して十分大きなサイズにするか、十分長い時間を経てからクラウドストレージに出力できます。ここで Flink はクラウドストレージのマルチパートアップロード (MPU) 機能を利用しています。各チェックポイントで Flink は現在のチェックポイントに保存されたデータをクラウドストレージにアップロードしますが、出力はファイルではなくパートです。最終的に、複数のパートがサイズまたは時間の条件を満たすと、クラウドストレージのインターフェイスを呼び出して複数のパートを 1 つのファイルに結合します。この結合操作はクラウドストレージ側で完了し、アプリケーション側でパートを再読み込みしてローカルで結合してからアップロードする必要はありません。バルク形式は一度に全体を処理する必要があるため、パート分割アップロードと結合はできず、一括でアップロードする必要があります。

2 つ目のジョブは新しい JSON ファイルがアップロードされたことを検知すると、そのファイルを読み込んで RCFile に変換し、最終パスにアップロードします。この処理による遅延はわずかであり、1 ファイルあたり 10 秒以内に抑えられ、許容範囲内です。

2. 入力ファイルのスマートな検知
入力側では、Flink の FileStreamingSource ではなく、クラウドストレージのイベント通知を利用して新しいファイルの生成を検出し、通知を受信してからファイルを能動的に読み込むようにしています。

3. パーティションの検知性と整合性
出力側では、dt レベルの success ファイルを出力して、下流が日次テーブルの準備完了を確実に検知できるようにします。カスタムの StreamingFileWriter を実装して partitionCreated と partitionInactive シグナルを出力し、これらのシグナルに基づいて日次テーブルの終了を判断するカスタムの PartitionCommitter を実装します。

仕組みは以下の通りです。各クラウドストレージライターは、特定の action の書き込みを開始すると partitionCreated シグナルを送信し、終了時に partitionInactive シグナルを送信します。PartitionCommitter は、特定の日付内のすべてのパーティションが inactive 状態かどうかを判断し、そうなっていればその日のすべてのデータが処理完了とみなして、dt レベルの success ファイルを出力します。Airflow はこのファイルを検知して、Flink が日次テーブルの処理を完了したかどうかを判断します。

4. Exactly Once
クラウドストレージのイベント通知は At Least Once の保証を提供します。Flink ジョブ内でファイルレベルの重複排除を行います。ジョブは Exactly Once のチェックポイント設定を採用しています。クラウドストレージへのファイル出力は MPU メカニズムに基づいており、切り捨てをサポートしているため、クラウドストレージへの出力は実質的に冪等です。したがって、エンドツーエンドの Exactly Once を実現しています。

7. プロジェクトの成果と展望

プロジェクトはすでに本番環境にリリースされており、レイテンシは約 34 分です。これには遅延ファイルの待機時間 15 分が含まれています。

最初の Flink ジョブのチェックポイントと出力の完了は約 8 分、JSON から RC への変換ジョブは 12 分で全処理を完了します。この時間をさらに短縮することも可能ですが、適時性とコストのバランスを考慮し、現在の状態を選択しています。

JSON から RC への変換ジョブは当初の予想以上に時間がかかっています。上流ジョブの最終チェックポイントで出力されるファイルが多すぎて、全体の処理時間が長期化しているのが原因です。これはジョブの同時実行数を増やすことで線形に短縮できます。

出力ファイル数はバッチジョブと比較して約 50% 増加しています。これはストリーム処理のバッチ処理に対する不利な点です。ストリーム処理では時間ウィンドウが終了した時点でファイルを出力する必要があり、その時点のファイルサイズが期待値に達していない場合があります。幸い、この程度のファイル数増加は下流のパフォーマンスに大きな影響を与えていません。

下流は完全に透過的で、リリース前後で異常なユーザーフィードバックは受けていません。
本プロジェクトを通じて、本番環境のバッチ処理システムにストリーム処理フレームワーク Flink をシームレスに組み込み、ユーザーが意識することなく部分的な改善を達成できることを実証しました。今後は同じ技術を活用して、より多くの Hive テーブルの生成を加速させ、時間レベルなどの細かい粒度の Hive テーブル生成にも幅広く対応していく予定です。一方で、データレイクを活用したバッチとストリーミングの統合データ管理を探索し、技術スタックの段階的な収束を実現していきます。