チュートリアル: StreamingQueryListener を使用した Structured Streaming クエリーの監視
StreamingQueryListener は、実行中の Structured Streaming クエリーから、各マイクロバッチの終了時に、入力行数、処理速度、バッチ期間などの進捗メトリクスを受け取ります。このチュートリアルでは、クエリーが入力に追いついているかどうかを確認し、クエリーが遅れた場合に警告を発し、各バッチの正常性を、SQLでクエリーできる Unity Catalog テーブルに記録するリスナーをアタッチする方法を説明します。この例は、Serverless コンピュート と Unity Catalog 有効化コンピュートで実行されます。
リスナーが報告するメトリクスとイベント・フィールドの完全なリファレンスについては、Databricks での Structured Streaming クエリーのモニタリングを参照してください。
このチュートリアルには、次のステップが含まれます。
- ノートブックを作成する
- Unity Catalogオブジェクトを作成します
- リスナーを定義する
- 定期的なライターを定義する
- モニタリングを使用してクエリーを起動する
- 記録された正常性をクエリーする
- クエリーを停止してクリーンアップする
要件
このチュートリアルを完了するには、以下の条件を満たす必要があります。
- ワークスペースはUnity Catalogが有効になっていなければなりません。Unity Catalog とはを参照してください。
- Serverless コンピュートまたは Unity Catalog 有効化クラスターにアクセスできる必要があります。
StreamingQueryListenerがサポートする Databricks Runtime バージョンとアクセスモードについては、Databricks での Structured Streaming クエリーのモニタリングを参照してください。 - Unity Catalog カタログにスキーマ、ボリューム、およびテーブルを作成する権限が必要です。このチュートリアルでは、
mainという名前のカタログを使用します。書き込み可能なカタログに置き換えます。
ステップ 1: ノートブックを作成する
Databricksノートブックを使用して、リスナーを定義し、ストリーミングクエリーをランします。
ノートブックを作成するには：
- サイドバーの [新規] をクリックし、
[ノートブック] をクリックします。 - コンピュートのドロップダウンメニューで、ServerlessコンピュートまたはUnity Catalog有効化クラスターを選択します。
- Python をdefault言語として選択します。
ステップ2: Unity Catalog オブジェクトを作成する
リスナーはバッチごとの正常性を Unity Catalog マネージドテーブルに記録し、ストリーミング クエリーは2番目のマネージドテーブルに書き込み、クエリーはそのチェックポイントを Unity Catalog ボリュームに保存します。スキーマとボリュームを作成し、オブジェクト名を定義するには、次をランします：
catalog = "main"
schema = "streaming_monitoring"
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog}.{schema}.checkpoints")
health_table = f"{catalog}.{schema}.streaming_health"
target_table = f"{catalog}.{schema}.rate_target"
checkpoint_path = f"/Volumes/{catalog}/{schema}/checkpoints/rate_target"
ステップ3: リスナーを定義する
リスナーは各コールバックを軽量に保ちます。マイクロバッチごとに、処理レートを入力レートと比較し、クエリーが遅延したときに警告を出力し、バッチの正常性をインメモリ・バッファに追加します。リスナーを定義するには、以下をランします。
import threading
from pyspark.sql.streaming import StreamingQueryListener
class HealthListener(StreamingQueryListener):
def __init__(self):
self._lock = threading.Lock()
self._buffer = []
def onQueryStarted(self, event):
print(f"Query started: {event.id}")
def onQueryProgress(self, event):
progress = event.progress
input_rate = progress.inputRowsPerSecond
processed_rate = progress.processedRowsPerSecond
# The query is falling behind when it processes data slower than it arrives.
falling_behind = input_rate > 0 and processed_rate < input_rate
if falling_behind:
print(f"WARNING batch {progress.batchId}: falling behind "
f"({processed_rate:.1f} processed/s < {input_rate:.1f} input/s)")
with self._lock:
self._buffer.append((int(progress.batchId), int(progress.numInputRows),
float(input_rate), float(processed_rate),
bool(falling_behind)))
def drain(self):
with self._lock:
records, self._buffer = self._buffer, []
return records
def onQueryIdle(self, event):
pass
def onQueryTerminated(self, event):
print(f"Query terminated: {event.id}")
リスナーは、次のイベントメソッドをオーバーライドします。
手法 | 説明 |
|---|---|
| クエリーが起動するときに実行されます。 |
| 各マイクロバッチの終了時にランします。この例では、処理レートを入力レートと比較し、クエリーが遅延したときに警告を発し、バッチの正常性をバッファリングします。 |
| クエリーがアイドル状態で新しいデータを待機しているときにランします。 |
| エラーの有無にかかわらず、クエリーが停止したときにランします。 |
drain メソッドは、バッファリングされたレコードを返し、バッファをクリアします。リスナーはリスナー スレッドからバッファに追加し、次のステップのライターは別のスレッドからバッファを読み取るため、ロックによってバッファが保護されます。
onQueryProgress リスナースレッドで実行されるため、そこでの負荷の高い作業は、セッション内のすべてのクエリーの進捗報告を遅らせます。これが、リスナーがコールバックでバッファリングのみを行い、別のスレッドから書き込む理由です。Databricks での Structured Streaming クエリーのモニタリング のメモを参照してください。
ステップ4：定期ライターを定義する
バックグラウンドスレッドはバッファーされたレコードを固定間隔で正常性テーブルに書き込むため、リスナーは独自のスレッドで Spark ジョブをランすることはありません。publish 関数は、レコードを書き込む唯一の場所です。メトリクスを Apache Kafka などの別のシステムに送信するには、publish の本文を置き換えます。ライターとその間隔ループを定義するには、次をランします：
def publish(records):
if not records:
return
(spark.createDataFrame(records,
"batchId long, numInputRows long, inputRowsPerSecond double, "
"processedRowsPerSecond double, fallingBehind boolean")
.write.mode("append").saveAsTable(health_table))
stop_event = threading.Event()
def flush_loop(listener, interval_seconds=10):
while not stop_event.wait(interval_seconds):
publish(listener.drain())
ループは、stop_eventが設定されるまで10秒ごとにレコードのバッチを書き込みます。書き込みごとに一度に多くの行が追加されるため、バッチごとの書き込みによって発生する可能性のある小さなファイルが回避されます。
ステップ5: モニタリング付きでクエリーを起動する
リスナーを登録し、ライター・スレッドを起動してから、組み込みレートソースから読み取ってターゲットテーブルに書き込むクエリーを起動します。レートソースは固定レートでデータ行を生成するため、リスナーは外部データソースを使用せずに安定した進捗状況を報告します。モニタリングとクエリーを起動するには、以下をランします。
listener = HealthListener()
spark.streams.addListener(listener)
flusher = threading.Thread(target=flush_loop, args=(listener,), daemon=True)
flusher.start()
query = (
spark.readStream.format("rate").option("rowsPerSecond", 5).load()
.writeStream.format("delta")
.option("checkpointLocation", checkpoint_path)
.toTable(target_table)
)
クエリーを1～2分ランして、いくつかフラッシュ間隔が経過するようにします。各マイクロバッチは onQueryProgress をTriggerし、クエリーが遅れた場合に警告を出力し、バッチの正常性をバッファーします。ライターはバッファーされたレコードを10秒ごとにテーブルに追加するため、ストリームがランしている間にクエリーを実行できます。
ステップ6：記録された正常性をクエリーする
正常性テーブルを読み取ると、各バッチで、データの到着速度、クエリーによる処理速度、クエリーの遅延状況を確認できます。レコードを表示するには、以下をランします。
display(spark.sql(f"""
SELECT batchId, numInputRows,
round(inputRowsPerSecond, 1) AS input_per_second,
round(processedRowsPerSecond, 1) AS processed_per_second,
fallingBehind
FROM {health_table}
ORDER BY batchId
"""))
前のクエリーは、完了したバッチごとに1行を返します。次に例を示します。
batchId numInputRows input_per_second processed_per_second fallingBehind
0 2 0.0 1.9 false
1 5 4.8 5.7 false
2 5 5.7 4.6 true
3 5 4.6 6.0 false
4 4 4.8 5.0 false
5 4 5.0 4.4 true
6 5 5.5 6.1 false
7 4 4.8 4.1 true
正常なクエリーは、入力データと少なくとも同じ速さでデータを処理するため、processed_per_second は input_per_second 以上に維持され、fallingBehind は false になります。バッチ2、5、7が示すように、個々のバッチは入力レートを下回ることがよくあります。このジッターは正常です。fallingBehind が連続するいくつかのバッチで true のままの場合、クエリーは入力に追いつくことができません。コンピュートを追加する、各バッチの作業を減らす、または入力レートを下げて調査します。
各プログレスイベントで利用可能なすべてのソース、シンク、ステータスメトリクスのセットについては、Databricks での Structured Streaming クエリーのモニタリング を参照してください。
ステップ 7: クエリーを停止してクリーンアップする
クエリーを停止し、ライター・スレッドを停止し、バッファー内に残っているレコードを書き込み、作成したオブジェクトをドロップして、コストが発生し続けないようにします。クリーンアップするには、次をランします。
query.stop()
stop_event.set()
flusher.join()
publish(listener.drain())
spark.streams.removeListener(listener)
spark.sql(f"DROP SCHEMA IF EXISTS {catalog}.{schema} CASCADE")