機能詳細解説：Apache Spark Structured Streaming におけるウォーターマーキング

主なポイント

• ウォーターマークは、イベント時刻に基づいて処理の進捗状況、ウィンドウ集計の生成タイミング、集計状態のトリミング時期をSparkが把握するのに役立ちます。
• データストリームを結合する際、Sparkはデフォルトで単一のグローバルウォーターマークを使用し、入力ストリーム全体で検出された最小イベント時刻に基づいて状態を削除します。
• RocksDBを活用することで、クラスターメモリへの負荷やGC（ガベージコレクション）の一時停止時間を短縮できます。
• StreamingQueryProgressおよびStateOperatorProgressオブジェクトには、ウォーターマークがストリームにどのように影響するかに関する重要な情報が含まれています。

はじめに

リアルタイムパイプラインを構築する際、分散データ取り込みは本質的に順序付けられていないという現実があります。さらに、ステートフルなストリーミング操作のコンテキストでは、時間ウィンドウ集計やその他のステートフルな操作を正しく計算するために、取り込んでいるデータストリームのイベント時刻の進捗を適切に追跡する必要があります。これらすべては、Structured Streamingを使用して解決できます。

たとえば、顧客にリースしている採掘機械のプロアクティブメンテナンスを行うためのパイプラインを構築しているチームを考えてみましょう。これらの機械は常に最高の状態で稼働する必要があるため、リアルタイムで監視します。機械の問題を理解し特定するために、ストリーミングデータに対してステートフルな集計を実行する必要があります。

ここで、Structured Streamingとウォーターマークを活用して、これらの機械の予知保全などに関する意思決定に役立つ必要なステートフル集計を生成する必要があります。

ウォーターマークとは？

一般的に、リアルタイムストリーミングデータを扱う場合、データの取り込み方法やアプリケーション全体でダウンタイムなどの問題が発生するかどうかにより、イベント時刻と処理時刻の間に遅延が生じます。これらの潜在的な変動遅延のため、このデータを処理するために使用するエンジンには、集計ウィンドウをいつ閉じて集計結果を生成するかを決定するメカニズムが必要です。

これらの問題を解決するために、壁時計時間に基づいた固定遅延を使用することが自然な考えかもしれませんが、この後の例で、なぜそれが最善の解決策ではないのかを示します。

視覚的に説明するために、午前10時50分から午前11時20分頃にかけてさまざまな時間にデータを受信するシナリオを考えてみましょう。10分間のタンブリングウィンドウを作成し、ウィンドウ期間中に受信した温度と圧力の読み取り値の平均を計算します。

最初の図では、午前11時、午前11時10分、午前11時20分にタンブリングウィンドウがトリガーされ、それぞれの時間に結果テーブルが表示されます。午前11時10分頃に2番目のデータバッチが、イベント時刻が午前10時53分のデータとともに到着すると、これは午前11時10分に閉じる午前11時00分〜午前11時10分のウィンドウの温度と圧力の平均に組み込まれますが、これは正しい結果をもたらしません。

生成したい集計の正しい結果を得るためには、Sparkが集計ウィンドウをいつ閉じて正しい集計結果を生成するかを理解できるようにするウォーターマークを定義する必要があります。

Structured Streamingアプリケーションでは、ウォーターマークと呼ばれる機能を使用して、計算したい集計に必要なすべてのデータが収集されるようにすることができます。最も基本的な意味では、ウォーターマークを定義することにより、Spark Structured Streamingは、設定された遅延予測に基づいて、時刻Tまでのすべてのデータが取り込まれたことを認識し、タイムスタンプTまでのウィンドウ集計を閉じて生成できます。

この2番目の図は、10分間のウォーターマークを実装し、Spark Structured StreamingでAppendモードを使用した効果を示しています。

最初のシナリオではSparkが10分ごとに前の10分間のウィンドウ集計を発行しますが（つまり、午前11時10分に午前11時00分〜午前11時10分のウィンドウを発行します）、Sparkは今や、検出された最大イベント時刻から指定されたウォーターマークを引いた値がウィンドウの上限よりも大きい場合にのみ、ウィンドウ集計を閉じて出力します。

言い換えると、Sparkは、検出された最新イベント時刻から10分を引いた値が午前11時よりも大きいデータポイントを見るまで待つ必要がありました。午前11時00分には、これは満たされないため、Sparkの内部状態ストアで集計計算のみが初期化されます。午前11時10分になってもこの条件は満たされませんが、午前10時53分の新しいデータポイントがあるため、内部状態は更新されますが、発行されません。そして最終的に午前11時20分までに、Sparkはイベント時刻が午前11時15分のデータポイントを見ており、11時15分から10分を引くと11時05分になり、これは11時00分よりも遅いため、午前10時50分〜午前11時00分のウィンドウが結果テーブルに発行できます。

これにより、ウォーターマークによって定義された予想される遅延に基づいてデータを適切に取り込むことで、正しい結果が得られます。結果が発行されると、対応する状態は状態ストアから削除されます。

パイプラインへのウォーターマークの組み込み

これらのウォーターマークをStructured Streamingパイプラインに組み込む方法を理解するために、このブログの導入セクションで述べたユースケースに基づいた実際のコード例をステップバイステップで見ていきます。

クラウド上のKafkaクラスターからすべてのセンサーデータを取得し、10分ごとに温度と圧力の平均を計算したいとします。予想される時間スキューは10分です。ウォーターマークを使用したStructured Streamingパイプラインは次のようになります。

PySpark

ここでは、Kafkaから読み取り、変換と集計を適用し、Delta Lakeテーブルに書き込みます。これはDatabricks SQLで視覚化および監視されます。テーブルに書き込まれた特定のデータサンプルの出力は次のようになります。

ウォーターマークを組み込むために、まず2つの項目を特定する必要がありました。

1. センサー読み取りのイベント時刻を表す列
2. データの推定される時間スキュー

前の例から、イベント時刻列としてeventTimestamp列を使用し、予想される時間スキューを表す10分間を指定した.withWatermark()メソッドによって定義されたウォーターマークを確認できます。

PySpark

Structured Streamingパイプラインにウォーターマークを実装する方法がわかったので、ストリーミングジョイン操作や状態管理などの他の項目がウォーターマークによってどのように影響を受けるかを理解することが重要になります。さらに、パイプラインをスケールアップするにつれて、パフォーマンスの問題を回避するためにデータエンジニアが認識し監視する必要がある主要なメトリックがあります。これらすべてを、ウォーターマークについてさらに詳しく見ていきます。

さまざまな出力モードでのウォーターマーク

さらに詳しく説明する前に、選択した出力モードが設定したウォーターマークの動作にどのように影響するかを理解することが重要です。

ウォーターマークは、ストリーミングアプリケーションをappendまたはupdate出力モードで実行している場合にのみ使用できます。3番目の出力モードであるcompleteモードでは、結果テーブル全体がストレージに書き込まれます。このモードは、すべての集計データを保持する必要があるため使用できず、したがって中間状態をドロップするためにウォーターマークを使用できません。

これらの出力モードがウィンドウ集計とウォーターマークのコンテキストで意味することは、「append」モードでは集計は一度しか生成できず、更新できないということです。したがって、集計が生成されると、エンジンはその集計の状態を削除できるため、全体的な集計状態を制限された状態に保つことができます。後続のレコード（近似ウォーターマークのヒューリスティックが適用されなかったレコード、つまりウォーターマーク遅延期間よりも古いレコード）は、集計が生成され、集計状態が削除されたため、必然的にドロップされる必要があります。

逆に、「update」モードでは、最初のレコードから開始して、受信した各レコードで集計を繰り返し生成できるため、ウォーターマークはオプションです。ウォーターマークは、エンジンが集計のためにこれ以上レコードを受信しないとヒューリスティックに判断できる場合にのみ、状態をトリミングするために役立ちます。状態が削除されると、集計値が失われ、更新できなくなったため、後続のレコードは再びドロップされる必要があります。

状態、遅延レコード、および異なる出力モードが、Sparkで実行されているアプリケーションの異なる動作につながる可能性があることを理解することが重要です。ここでの主なポイントは、appendモードとupdateモードの両方で、ウォーターマークが集計時間ウィンドウのすべてのデータが受信されたことを示すと、エンジンはウィンドウ状態をトリミングできるということです。appendモードでは、集計は時間ウィンドウの終了時プラスウォーターマーク遅延時にのみ生成されますが、updateモードではウィンドウのすべての更新時に生成されます。

最後に、ウォーターマーク遅延ウィンドウを増やすと、パイプラインはデータのために長く待機し、潜在的にドロップされるデータが少なくなるため、精度は高くなりますが、集計を生成するまでのレイテンシも高くなります。逆に、ウォーターマーク遅延が小さいと、精度は低くなりますが、集計を生成するまでのレイテンシも低くなります。

ウォーターマークの詳細

ジョインとウォーターマーク

ストリーミングアプリケーションでジョイン操作を行う場合、特に2つのストリームをジョインする場合、注意すべき点がいくつかあります。ユースケースとして、温度と圧力の読み取り値に関するストリーミングデータセットを、マシン全体で他のセンサーによってキャプチャされた追加の値とジョインしたいとします。

Structured Streamingで実装できるストリーム-ストリームジョインには、主に3つのタイプがあります。内部ジョイン、外部ジョイン、セミジョインです。ストリーミングアプリケーションでジョインを行う主な問題は、ジョインの片側の情報が不完全になる可能性があることです。Sparkに将来の照合がないことを理解させることは、集計を放出する前に集計の計算に組み込む新しい行がないことをSparkが理解する必要があった、集計に関する以前の問題と同様です。

Sparkがこれを処理できるように、ウォーターマークとイベントタイム制約をストリーム-ストリームジョインのジョイン条件内で組み合わせて利用できます。この組み合わせにより、Sparkは遅延レコードをフィルタリングし、ジョイン操作の状態を時間範囲条件でトリミングできます。以下に例を示します。

PySpark

ただし、上記の例とは異なり、各ストリームで異なる時間スキューのウォーターマークが必要になる場合があります。このシナリオでは、Sparkには複数のウォーターマーク定義を処理するためのポリシーがあります。Sparkは、データを見逃さないように最大限の安全性を確保するために、最も遅いストリームに基づいたグローバルウォーターマークを維持します。

開発者は、spark.sql.streaming.multipleWatermarkPolicyをmax;に変更することでこの動作を変更できますが、これは遅いストリームからのデータがドロップされることを意味します。

ウォーターマークを必要とする、または利用できるジョイン操作の全範囲を確認するには、Sparkのドキュメントのこのセクションを参照してください。

ウォーターマークによるストリームの監視と管理

Sparkが数百万のキーを管理し、それぞれに状態を保持する必要があるストリーミングクエリを管理する場合、Databricksクラスターに付属するデフォルトの状態ストアでは効果がない場合があります。メモリ使用量が増加し、ガベージコレクションの一時停止時間が長くなる可能性があります。これらは両方とも、Structured Streamingアプリケーションのパフォーマンスとスケーラビリティを妨げます。

ここでRocksDBが登場します。Spark構成で次のように有効にすることで、DatabricksでRocksDBをネイティブに利用できます。

これにより、Structured Streamingアプリケーションを実行しているクラスターはRocksDBを利用できるようになり、ネイティブメモリの状態をより効率的に管理し、すべてをメモリに保持するのではなく、ローカルディスク/SSDを活用できます。

メモリ使用量とガベージコレクションメトリックの追跡以外にも、ウォーターマークとStructured Streamingを扱う際に収集および追跡すべき他の重要な指標とメトリックがあります。これらのメトリックにアクセスするには、StreamingQueryProgressおよびStateOperatorProgressオブジェクトを確認してください。これらの使用方法の例については、こちらのドキュメントをご覧ください。

StreamingQueryProgressオブジェクトには、「eventTime」というメソッドがあり、これを呼び出すと、トリガーにおけるイベントタイムの最大値、最小値、平均値、およびウォーターマークタイムスタンプが返されます。最初の3つは、そのトリガーで見たイベントタイムの最大値、最小値、平均値です。最後の1つは、トリガーで使用されるウォーターマークです。

StreamingQueryProgressオブジェクトの省略された例

これらの情報は、ストリーミングクエリが出力する結果テーブルのデータを照合するために使用でき、また、使用されているウォーターマークが意図したイベントタイムスタンプであることを確認するためにも使用できます。これは、データストリームを一緒にジョインしている場合に重要になります。

StateOperatorProgressオブジェクト内には、numRowsDroppedByWatermarkメトリックがあります。このメトリックは、ステートフル集計に含めるには遅すぎると見なされる行数を示します。このメトリックは、生の入力行ではなく、集計後の行を測定していることに注意してください。したがって、数値は正確ではありませんが、遅延データがドロップされていることを示すことができます。これは、StreamingQueryProgressオブジェクトからの情報と組み合わせて、開発者がウォーターマークが正しく構成されているかどうかを判断するのに役立ちます。

複数の集計、ストリーミング、およびウォーターマーク

Structured Streamingクエリの残りの1つの制限は、単一のストリーミングクエリで複数のステートフルオペレーター（例：集計、ストリーミング結合）を連鎖させることです。ステートフル集計用の単一のグローバルウォーターマークというこの制限は、Databricksで解決策に取り組んでおり、今後数か月でさらに詳しい情報を公開する予定です。詳細については、Project Lightspeedに関するブログをご覧ください： Project Lightspeed: Apache Sparkによる、より高速でシンプルなストリーム処理 (databricks.com)。

結論

Databricks上のStructured StreamingとWatermarkingを使用すると、上記の使用例のような組織は、データが適切に並べ替えられていない、またはオンタイムでない場合でも、リアルタイム集計によって駆動されるメトリックが正確に計算されることを保証する、回復力のあるリアルタイムアプリケーションを構築できます。Databricksでリアルタイムアプリケーションを構築する方法の詳細については、Databricksの担当者にお問い合わせください。

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