From Spark for batch processing to Flink for streaming and batch integration

1. ストリームバッチ統合の必要性

ストリームとバッチの統合にはどのようなメリットがあるのでしょうか。特に BI / AI / ETL の文脈で考えてみます。全体として、ストリームとバッチの統合を実現できれば、以下の 4 つの明確なメリットが得られます。

コードの重複を回避し、コア処理ロジックを再利用できる

コードロジックを完全に一致させることができれば理想的ですが、現実的には難しい面もあります。しかし全体として、現在のビジネスロジックはますます長く複雑になっており、多くの要件が存在します。異なるフレームワークや異なるエンジンを使用している場合、ユーザーは毎回ロジックを書き直す必要があり、非常に大きな負担となり、保守も困難になります。そのため、コードの重複をできる限り回避し、ユーザーがコードロジックを再利用できるようにすることは特に重要です。

ストリームバッチには 2 つの方向性がある

この 2 つの方向性で考慮すべき課題は大きく異なります。現在、ストリーム処理向けの Flink やバッチ処理向けの Spark といったフレームワークは、それぞれの処理領域で比較的成熟しており、多くのユーザーを獲得しています。ユーザーを別の方向性へ移行させる場合、たとえばストリーム処理からバッチ処理へ移行するか、バッチ処理からストリーム処理へ移行するかという 2 つのカテゴリーに分けられます。後述する 2 つの本番事例は、それぞれこの 2 つの方向性に対応しています。

メンテナンス負荷の軽減

複数システムの保守を回避できます。システム間の差異は非常に大きく、フレームワークやエンジンも異なるため、より多くの問題が生じます。リアルタイム用とオフライン用のパイプラインが社内に別々に存在すると、データの不整合が発生します。そのため、データ検証、データ精度の照会、データストレージなどでデータ整合性を可能な限り保つために、多くの作業が必要になります。

詳細を見る

フレームワークやエンジンが多く存在し、ビジネスロジックはリアルタイムとオフラインの両方で稼働する必要があります。そのため、ユーザーをサポートするには多くの学習が必要となります。

2. 業界の現状

Flink と Spark はいずれもストリーム処理とバッチ処理の両方をサポートするエンジンです。Flink のストリーム処理の方が優れているという点では合意していますが、そのバッチ処理はどの程度優れているのでしょうか。同時に、Spark のバッチ処理は比較的優れていますが、そのストリーム処理はユーザーの既存のニーズを十分に解決できるのでしょうか。

現在、さまざまなエンジンフレームワークが存在しますが、その上に統合フレームワークや、Beam API やカスタムインターフェースのようなシンプルな物理 API に似たものを構築できるのでしょうか。

Beam で検討すべき課題は、バッチ処理とストリーム処理をどこまで最適化できるかということです。Beam は現在まだ物理実装に偏っており、今後のロードマップを研究する必要があります。

LinkedIn を含む他の企業は、共通 SQL レイヤーや共通 API レイヤーを設け、その下層で異なるフレームワークエンジンを稼働させるカスタムインターフェースソリューションを検討しています。ここで検討すべき課題は、Spark や Flink のようなフレームワークは既に比較的成熟しており、大規模なユーザー基盤を持っているということです。新しい API や新しいソリューションを提案する際、ユーザーの受け入れ度はどの程度か。新しいソリューションは社内でどのように保守すべきか。

3. 本番事例シナリオ

以下の内容は、主に Flink のバッチ処理としての効果、Flink と Spark の簡単な比較、および LinkedIn の社内ソリューションに焦点を当てています。2 つの本番事例シナリオを紹介します。1 つ目は機械学習の特徴量エンジニアリング生成時におけるストリームバッチ統合の方法、2 つ目は複雑な ETL データフローにおけるストリームバッチ統合の方法です。

3.1 Case A - 機械学習の特徴量エンジニアリング

1 つ目の方向性であるストリーム処理 → バッチ処理は、ストリームバッチ統合に分類されます。

Case A の主なロジックは、機械学習の特徴量生成時にストリーム処理からバッチ処理へのストリームバッチ統合を行う方法です。コアとなるビジネスロジックは特徴量変換で、変換プロセスとロジックが複雑なため、標準化の事例として取り上げます。

たとえば、LinkedIn ページに入力されたメンバーのプロフィール情報などのバックグラウンドを抽出・標準化し、求人の推奨などに活用するケースがあります。メンバーの ID 情報が更新されると、フィルタリングと前処理のロジックが実行されます。これには Kafka の読み取りプロセスも含まれ、特徴量変換中に小規模なテーブルクエリが発生することもあります。このロジックは非常にシンプルで、複雑な join 操作やその他のデータ処理プロセスは含まれません。

以前のパイプラインはリアルタイムで、オフラインパイプラインから補足情報を読み取ってストリームを定期的に更新する必要がありました。このバックフィルがリアルタイムクラスターに大きな負荷をかけていました。バックフィル中はジョブの完了を待機し、リアルタイムクラスターがダウンしないようワークフローを監視する必要がありました。そのため、ユーザーからリアルタイムストリーム処理ではなくオフラインでバックフィルを行いたいという要望がありました。

現在のユーザーはストリーム処理に Samza 上の Beam を使用しています。Beam API と Spark Dataset API に精通しており、バックフィル以外の業務処理でも Dataset API を使用しています。

ここで強調すべき点は、多くの Dataset API はオブジェクトを直接操作し、型安全性に対する要件が高いことです。これらのユーザーに SQL や DataFrame のワークフローへ直接移行することを提案するのは現実的ではありません。既存のビジネスロジックはオブジェクトを直接操作・変換するものだからです。

このような場合、ユーザーに Imperative API の選択肢を提供できます。業界が提供するソリューションを見てみましょう。

1 つ目の選択肢は Flink DataStream API で、統一向けに開発が進められています。過去のプログラム評価時には Flink DataSet API（非推奨）も調査しました。DataStream API は統合でき、ストリーム処理とバッチ処理のサポートも比較的良好です。ただし、Imperative API であるため最適化の余地は限られており、今後も継続的に最適化されていく見込みです。FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API) および FLIP-134: Batch execution for the DataStream API を参照してください。

2 つ目の選択肢は Spark Dataset で、これもユーザーにとって自然な選択肢です。Dataset API はストリーミングにも使用できますが、これは Flink の Dataset や DataStream API などの物理 API とは異なります。Spark Dataframe SQL エンジンをベースに型安全性を実現しており、最適化の度合いは比較的良好です。Databricks: Introducing Apache Spark Datasets および Spark Structured Streaming Programming Guide: Unsupported-operations の記事を参照してください。

3 つ目の選択肢は Beam on Spark で、現在は主に RDD ランナーを使用しています。最適化を備えたランナーのサポートは依然として難しい状況です。Case B で Beam の現在進行中の作業について詳しく説明します。Beam Documentation - Using the Apache Spark Runner および BEAM-8470 Create a new Spark runner based on Spark Structured streaming framework を参照してください。

ユーザーのフィードバックによると、Flink の DataStream (DataSet) API と Spark の Dataset API はユーザーインターフェースの面で非常に近いとのことです。インフラエンジニアとしてユーザーの問題を解決するには、API への習熟度の方がより重要です。

ただし、Beam の API は Flink や Spark とは大きく異なります。Beam は Google のエコシステムに属しています。以前、ユーザーの問題解決を支援した際、彼らのワークフローは Samza 上の Beam で、PCollection や PTransform を使ってビジネスロジックを記述していました。出力と入力のメソッドシグネチャが大きく異なるため、軽量なコンバーターを開発し、既存のビジネスロジックを Flink や Spark のジョブで再利用できるようにしました。

DAG の観点から見ると、Case A はオブジェクトを単純かつ直接的に変換する非常にシンプルなビジネスプロセスです。この場合、Flink と Spark のパフォーマンスは非常に近くなります。

通常、Flink のダッシュボード UI を使って例外やビジネスプロセスなどを確認しますが、これは Spark に比べて明らかな優位性です。Spark ではドライバーログを照会して例外を調査する必要があり、より手間がかかります。ただし、Flink にはまだ改善が必要な領域がいくつかあります。

History Server - より豊富なメトリクスなどのサポート

Spark History Server UI が提示するメトリクスは比較的豊富で、パフォーマンス分析においてユーザーにとって大きな助けとなります。Flink のバッチ処理でも同程度のメトリクス情報を利用できるようにすれば、ユーザーの開発難易度を下げ、開発効率を向上できます。

より優れたバッチ運用保守ツール

LinkedIn が 2〜3 年前から取り組んでいる取り組みを共有します。LinkedIn では毎日 20 万件のジョブがクラスター上で稼働しており、バッチユーザーが自身のジョブをより効率的に運用・保守できるよう、より良いツールが必要です。Dr. Elephant と GridBench を提供し、ユーザーのジョブデバッグと運用を支援しています。

Dr. Elephant はオープンソースで、ユーザーがジョブをより効率的にデバッグし、問題を特定して改善提案を行うことができます。また、テストクラスターから本番クラスターへ移行する際、Dr. Elephant が生成する評価レポートのスコアに基づいて本番投入の可否を判断します。

GridBench は主に CPU メソッドのホットスポット分析などのデータ統計分析を行い、ユーザーのジョブ最適化と改善を支援します。GridBench は将来的にオープンソース化される予定で、Flink を含むさまざまなエンジンフレームワークをサポートし、GridBench を使った Flink ジョブのより優れた評価が可能になる見込みです。GridBench Talk: Project Optimum: Spark Performance at LinkedIn Scale を参照してください。

ユーザーは GridBench と Dr. Elephant が生成するレポートだけでなく、コマンドラインからもジョブの基本的な情報（アプリケーションの CPU 時間、リソース消費など）を確認でき、異なる Spark ジョブと Flink ジョブの比較分析も実施できます。

以上が Flink のバッチ処理で改善が必要な 2 つの領域です。

3.2 Case B - 複雑な ETL データフロー

2 つ目の方向性であるバッチ処理 → ストリーム処理は、ストリームバッチ統合に分類されます。

ETL データフローのコアロジックは比較的複雑で、たとえばセッションウィンドウ集計やウィンドウ集計が含まれ、1 時間ごとのユーザーページビューを計算し、異なるジョブに分割して、中間メタデータテーブルでページキーを共有します。00 時のジョブと 01 時のジョブでセッション化の処理を行い、最終的にオープンセッションとクローズセッションに分けて出力することで、各時間帯のデータを増分処理します。

このワークフローは元々 Spark SQL を使ったオフラインの増分処理で、純粋なオフライン増分処理でした。ユーザーがジョブをオンラインに移行してリアルタイム処理を行いたい場合、Samza 上の Beam などのリアルタイムワークフローを新たに構築する必要があります。構築過程で、ユーザーと密接に連携・コミュニケーションを取りましたが、多くの問題に直面しました。開発ロジック全体の再利用、2 つのビジネスロジックが同じ結果を生成することの保証、データの最終保存先などです。移行には長い時間を要し、最終的な効果はあまり芳しくありませんでした。

さらに、ユーザーのジョブロジックでは Hive と Spark の両方で多くの大規模かつ複雑な UDF が記述されており、この移行も非常に重い作業でした。ユーザーは Spark SQL と Spark DataFrame API に精通しています。

上図の黒い実線はリアルタイム処理、グレーの矢印は主にバッチ処理を示しており、Lambda アーキテクチャに相当します。

Case B のジョブには多数の join とセッションウィンドウが含まれており、ユーザーは以前から Spark SQL を使って開発していました。明らかに Declarative API から始める必要があります。現在 3 つのソリューションが提供されています。

1 つ目の選択肢は Flink Table API/SQL で、ストリーム処理とバッチ処理の両方を実行できます。同一の SQL で包括的な機能サポートがあり、ストリーム処理とバッチ処理の両方に最適化も施されています。Alibaba Cloud Blog: What's All Involved with Blink Merging with Apache Flink？ および FLINK-11439 INSERT INTO flink_sql SELECT * FROM blink_sql の記事を参照してください。

2 つ目の選択肢は Spark DataFrame API/SQL で、これも同一のインターフェースでバッチ処理とストリーム処理に使用できますが、Spark のストリーム処理サポートはまだ十分とは言えません。

3 つ目の選択肢は Beam Schema Aware API/SQL です。Beam は物理 API の色が強く、Schema Aware API/SQL の初期段階の作業が現在進行中のため、当面は検討対象外とします。したがって、以降の主な分析結果と経験は Flink Table API/SQL と Spark DataFrame API/SQL の比較から得られたものです。Beam Design Document - Schema-Aware PCollections および Beam User Guide - Beam SQL overview の記事を参照してください。

ユーザーの観点から見ると、Flink Table API/SQL と Spark DataFrame API/SQL は非常に近いです。キーワード、ルール、join の記述方法など、比較的小さな違いがありますが、これらはユーザーにある程度の混乱をもたらす可能性があります。正しく使っているのかどうか気になってしまうでしょう。

Flink と Spark はどちらも Hive との統合が進んでおり、Hive UDF の再利用なども可能なため、Case B の UDF 移行にかかる負担を半減させています。

Flink のパイプラインモードでのパフォーマンスは Spark よりも明らかに優れています。ディスクへの書き込みが発生するかどうかはパフォーマンスに大きな影響を与えることは想像に難くありません。大量のディスク書き込みが必要な場合、バッチモードでは各ステージでデータをディスクに書き込む必要があり、読み取り直しのオーバーヘッドがあるため、処理パフォーマンスはディスクに書き込まないパイプラインモードよりも劣ります。パイプラインは短時間の処理に適しており、20 分から 40 分の処理時間ではまだ比較的大きな優位性があります。パイプラインが長くなると、フォールトトレランスの面でバッチモードに劣ります。Spark のバッチパフォーマンスは依然として Flink よりも優れており、この領域は社内のケースに基づいて評価する必要があります。

Flink のウィンドウサポートは他のエンジンよりも明らかに充実しており、たとえばセッションウィンドウはユーザーにとって非常に使いやすい機能です。以前、ユーザーはセッションウィンドウを実現するために多くの UDF を作成する必要がありましたが、現在ではセッションウィンドウオペレーターを直接使用することで、開発コストを大幅に削減できています。同時に、グループ集約などのウィンドウ操作もストリームバッチでサポートされています。

UDF はエンジンフレームワーク間の移行における最大のハードルです。UDF が Hive で記述されていれば移行は容易です。Flink と Spark はどちらも Hive UDF を非常に良くサポートしているからです。しかし、UDF が Flink や Spark で記述されている場合、他のエンジンフレームワークへの移行は非常に大きな問題に直面します。たとえば、準リアルタイム OLAP クエリのために Presto へ移行する場合などです。

UDF の再利用を実現するために、LinkedIn では Transport プロジェクトを社内開発し、オープンソースとして GitHub に公開しています。LinkedIn のブログ Transport: Towards Logical Independence Using Translatable Portable UDFs を参照してください。

Transport はすべてのエンジンフレームワークに対してユーザー向けの User API を提供し、共通の関数開発インターフェースを通じて、Presto、Hive、Spark、Flink などのエンジンフレームワークに応じた UDF を自動生成します。

共通の UDF API でエンジンフレームワークを接続することで、ユーザーは自身のビジネスロジックを再利用できます。ユーザーは簡単に使用でき、たとえば以下のユーザーが開発する

ユーザーの SQL 移行の問題に対処するため、以前は Spark SQL を使ってジョブを開発していましたが、ストリームバッチ統合を使用して Flink SQL に変更したいという要件がありました。現在も多くのエンジンフレームワークが存在します。LinkedIn は Coral ソリューションを社内開発し、オープンソースとして GitHub に公開しています。Facebook でも講演を行っており、Transport UDF と共に分離レイヤーを提供することで、ユーザーがクロスエンジンでの移行とビジネスロジックの再利用をより良く実現できるようにしています。

Coral の実行プロセスを見てみましょう。まず、使い慣れた ASCII SQL とテーブル属性をジョブスクリプトで定義し、Coral IR 木構造を生成して、最終的に各エンジンの物理プランに変換されます。

Case B の分析ではストリームとバッチが統合されています。クラスターの業務量が特に多い場合、ユーザーはバッチ処理のパフォーマンス、安定性、成功率を重視します。その中でも Shuffle Service はバッチ処理のパフォーマンスに大きな影響を与えます。

4. Spark と Flink の Shuffle Service の比較

In-memory Shuffle は Spark と Flink の両方でサポートされており、より高速ですが、スケーラビリティはサポートしていません。

Hash-based Shuffle も Spark と Flink の両方でサポートされています。In-memory Shuffle と比較してフォールトトレランスのサポートは優れていますが、こちらもスケーラビリティはサポートしていません。

Sort-based Shuffle は大規模な Shuffle に対応するスケーラビリティをサポートしています。ディスクから少しずつ読み取ってソートマッチを実行してから読み戻します。FLIP-148: Introduce Sort-Based Blocking Shuffle to Flink でもサポートされています。

External Shuffle Service は、クラスターが非常に混雑している時、たとえば動的リソーススケジューリング時などに非常に重要です。Shuffle のパフォーマンスとリソース依存性の分離がより良く、分離後のリソーススケジューリングも効率的に行えます。FLINK-11805 A Common External Shuffle Service Framework は現在再開中です。

Disaggregate Shuffle について、ビッグデータ分野ではクラウドネイティブが提唱されており、計算とストレージの分離も Shuffle Service の設計で考慮する必要があります。FLINK-10653 Introduce Pluggable Shuffle Service Architecture では、プラグイン可能な Shuffle Service アーキテクチャが導入されています。

Spark は Shuffle Service に比較的大きな改善を加えました。この作業は LinkedIn が主導する Magnet プロジェクトでもあり、Magnet: A scalable and performant shuffle architecture for Apache Spark という論文としてまとめられ、2020 年の LinkedIn ブログに掲載されました。Magnet はディスク読み書きの効率を明らかに向上させます。小規模なランダム読み取りから大規模なシーケンシャル読み取りへと移行し、ランダムな読み取りではなくマージ処理を行うことで、ランダム I/O の問題を回避しています。

Shuffle の安定性とスケーラビリティの問題は Magnet Shuffle Service によって緩和されました。以前はジョブの失敗率が高いなど、多くの Shuffle の問題が発生していました。Flink をバッチ処理に使用し、以前 Spark でバッチ処理を行っていたユーザーを支援するには、Shuffle により多くの労力を費やす必要があります。

Shuffle の可用性については、best-effort 方式で Shuffle ブロックをプッシュし、一部の大きなブロックをスキップして結果整合性と精度を保証します。
Shuffle の一時データのコピーを作成して精度を保証します。
プッシュプロセスが特に遅い場合は、早期終了機構が適用されます。

Vanilla Shuffle と比較して、Magnet Shuffle は Shuffle データの読み取り待機時間をほぼ 100% 削減し、タスク実行時間をほぼ 50% 削減、エンドツーエンドのタスク持続時間をほぼ 30% 削減しました。

5. まとめ

LinkedIn は、Flink がストリーム処理とバッチ処理において明らかな優位性を持ち、より統合的で継続的な最適化が行われていることを認識し、喜ばしく思っています。

Flink のバッチ処理機能には改善が必要です。History Server、メトリクス、デバッグなどがその例です。ユーザーは開発時にユーザーコミュニティのソリューションを参照する必要があり、エコシステム全体を構築してユーザーが便利に使えるようにする必要があります。

Flink は Shuffle Service と大規模クラスターのオフラインワークフローにより多くの投資を行う必要があり、ワークフローの成功率を保証し、スケーラビリティが増す中でユーザーサポートとクラスターの健全性監視をどのように提供するかが課題です。

フレームワークエンジンが増えるにつれ、ユーザーにより統合されたインターフェースを提供することが望ましいです。この領域の課題は比較的大きく、開発と運用保守が含まれます。LinkedIn の経験によると、依然として多くの問題が見られ、単一のソリューションですべてのユーザーのユースケースをカバーすることは不可能であり、一部の機能や表現を完全にカバーすることも困難です。Coral や Transport UDF のように。