Databricks 上の Spark Declarative Pipelines 向けリアルタイムモードの一般提供 (GA) の開始

昨年、低遅延のオペレーショナルなユースケースの構築を可能にする Real-Time Mode (RTM) を発表しました。これにより、従来のマイクロバッチアーキテクチャと比較して最大100倍高速なミリ秒単位でのストリーミングデータの処理が可能になり、大きな変革をもたらしました。さらに、Spark と並行して Flink のような別個の専用エンジンを維持する必要がなくなり、重複するコードベース、断片化されたガバナンスモデル、運用上のオーバーヘッドが解消されました。

本日、Spark Declarative Pipelines (SDP) における RTM が一般提供 (GA) 開始されたことを発表いたします。SDP は Databricks 上でストリーミングおよび ETL パイプラインを構築するための標準的な手法となり、依存関係の解決、インクリメンタル処理、オーケストレーション、障害からの復旧、データ品質の適用を自動的に処理します。RTM を使用することで、それらの SDP のメリットをすべて維持しながら、オペレーショナルで低遅延のパイプラインを構築できます。不正行為のスコアリング、リアルタイムのアラート、オペレーショナルストアへの最新特徴量のプッシュなど、あらゆるユースケースで SDP をご利用いただけます。

Real-Time Mode (RTM) とは

RTM は、ミリ秒単位のエンドツーエンドの遅延を実現する Spark Structured Streaming エンジンの進化形です。データを継続的にストリーミングし、すべてのクエリステージを同時に実行し、データが生成され次第ステージ間で渡すことでこれを実現しています。アーキテクチャの詳細については、ブログ記事「マイクロバッチの壁を破る：Apache Spark Real-Time Mode のアーキテクチャ」をご覧ください。

RTM は、データをミリ秒単位で意思決定に変換するメリットがあるあらゆるワークロードをサポートします。ユースケースの例としては、不正検出、ライブパーソナライゼーション、リアルタイム推薦システム、リアルタイム特徴量計算などが挙げられます。実際のお客様のワークロードによる特徴量計算において、RTM と Apache Flink のベンチマークを比較したところ、RTM は Flink などの専用エンジンよりも最大 92% 高速であることが示されました。ベンチマークの結果は以下の通りです。参照されているデータセットとクエリの詳細については、こちらの GitHub リポジトリをご覧ください。

Spark Declarative Pipelines で RTM を使用する理由

お客様は ETL ワークロードにおいて SDP をデフォルトで利用する傾向が強まっています。RTM により、SDP はオペレーショナルな低遅延ストリーミングを含むすべてのストリーミングワークロードをサポートし、すべてのパイプラインで単一のフレームワークを使用できるようになります。単一のフレームワークとは、1つのオブザーバビリティスタック、1つの課金モデルを意味し、異なる API 間での断片化がなくなります。

SDP サーバーレスにより、そのメリットがさらに広がります。選択すべきコンピュートも、サイズを決定すべきクラスターもありません。複雑なインフラのチューニングを行わずにサブ秒の遅延を実現し、クエリが使用するコアに対してのみ料金が発生します。 また、RTM フローには、バージョン不要のアップグレードや稼働停止時間の少ないメンテナンスといった SDP の機能も備わっており、運用作業とアップグレードによる中断の両方を軽減します。

SDP上の RTM がすぐに効果を発揮するユースケースの1つが、リアルタイム特徴量計算です。不正検出やパーソナライゼーションのための ML モデルは、正確な予測を行うために、ユーザーが過去10分間に行ったトランザクションの合計金額などの最新のシグナルに依存しています。このようなローリングウィンドウ集計を計算し、ミリ秒単位でオンラインストアに書き込むという点で、SDP 上の RTM がそのギャップを埋めます。Databricks Feature Store はまさにこのアーキテクチャを使用して、Kafka からオンライン特徴量ストアまで、エンドツーエンドの p99 遅延 200ms を実現しています。

SDP での RTM の使用開始

ステップ 1: パイプラインの設定

Real-Time Mode では、パイプラインモードを continuous に設定し、spark.databricks.streaming.realTimeMode.enabled を true に設定して引き継ぐ必要があります。

ステップ 2: リアルタイム更新フローの定義

リアルタイムモードには更新フローが必要です。dp.create_sink() を使用して出力ターゲットを定義し、@dp.update_flow を pipelines.trigger に設定し、シンクを指す "RealTime" と target デコレータを使用します。

サポートされているソースとシンクについては、ドキュメントをご覧ください。

カスタムシンクへの特徴量の書き込み

不正検出やリアルタイムパーソナライゼーションなどのユースケースでは、特徴量がリアルタイムで計算されるだけでなく、クエリ可能な低遅延ストアに書き込まれる必要があります。不正検出モデルは決済が行われる前に顧客の最新のトランザクションを把握する必要があり、パーソナライゼーションエンジンはユーザーが閲覧したばかりのコンテンツを読み取る必要があります。

現在、Lakebase、Redis、またはその他の OLTP ストアなどのカスタムシンクへ、ForEach シンクを使用して SDP から書き込むことができます。ライターを定義し、foreach_sink として登録し、update_flow に接続します。

特徴量は生成から数ミリ秒以内にオペレーショナルストアに書き込まれ、その間にオフラインからオンラインへの同期は発生しません。アプリケーションやモデルから超低遅延でクエリを実行できます。
サーバーレスにおける RTM の今後の展開について

• RTM パイプラインの低コスト化: 各クエリが独自の資源を確保するのではなく、複数の RTM フローを単一のパイプラインにビンパックします。また、ワークロードに基づいてソースおよびシャッフルパーティションが自動チューニングされるため、ユーザーが手動でサイジングを行う必要がなくなります。
• シンクのサポート拡大: バッチ処理、自動再試行、UC 管理認証を内蔵した、SDP 用のネイティブ Lakebase シンク。

まとめ

RTM は、Spark Declarative Pipelines を、ストリーミングデータへの即座の応答が求められる、運用上の遅延に敏感な新しいクラスのワークロードへと拡張します。チームがすでに使用している SDP API にサブ秒の遅延をもたらすことで、最も時間的制約の厳しいパイプラインのために別のフレームワークやエンジンを運用する必要がなくなります。不正検出パイプライン、パーソナライゼーションエンジン、ML 特徴量計算システムのいずれを構築する場合でも、リアルタイムモードにより、SDP のシンプルさを維持しながらアプリケーションに必要な低遅延を実現できます。

Spark Declarative Pipelines における Real-Time Mode は現在、一般提供 (GA) 開始されています。

技術リソース

SDP での RTM の使用を開始するには、以下のリソースをご覧ください。

• ドキュメント: Lakeflow パイプラインでの Real-Time Mode の使用
• ドキュメント: Lakebase ストリーミングシンクへの接続
• ドキュメント: ForEach シンクを使用したパイプラインでの各行の書き込み
• コード例: SDP での Real-Time Mode のコード例

よくある質問 (FAQ)

Spark Declarative Pipelines の Real-Time Mode (RTM) とは何ですか?

リアルタイムモード (RTM) により、Spark Declarative Pipelines (SDP) はミリ秒単位のエンドツーエンドの遅延でストリーミングデータを処理できるようになります。データを継続的にストリーミングし、すべてのクエリステージを同時に実行し、データが生成され次第ステージ間で渡します。リアルタイムパイプラインは、SDP のオーケストレーションと自動障害復旧のメリットを享受できます。

Real-Time Mode は Spark Declarative Pipelines で利用できますか?

はい。Spark Declarative Pipelinesのリアルタイムモードが一般提供開始されました。Spark Structured Streamingですでに利用可能なリアルタイムモードがSDPパイプラインでも使えるようになります。

パイプラインでリアルタイムモードを有効にするにはどうすればよいですか？

パイプラインを continuous モードに設定し、spark.databricks.streaming.realTimeMode.enabled を true に設定します。次に、pipelines.trigger を "RealTime" に設定した更新フローと、シンクを指すターゲットを定義します。

リアルタイムパイプラインからLakebaseやその他のOLTPストアに書き込むことはできますか？

はい。open、process、close を実装するライターを持つForEachシンクを定義し、そこにリアルタイムフローを指定します。これにより、Lakebase、Redis、その他のOLTPストアで動作します。

Spark Declarative Pipelinesのリアルタイムモードはどのようなユースケース向けに構築されていますか？

リアルタイムモードは、ストリーミングデータをミリ秒単位で意思決定に活用するオペレーショナルワークロード向けに構築されています。例として、不正検出、ライブパーソナライゼーション、リアルタイムレコメンデーション、リアルタイム特徴量計算などが挙げられます。Databricksの特徴量ストアでは、このアーキテクチャを使用して、Apache Kafkaからオンライン特徴量ストアまでのp99レイテンシ200msを実現しています。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事