謝辞のコールバック

確認応答コールバックは、クライアントがプロデューサーループをブロックすることなく、非同期で確認応答やエラーを記録できるようにします。 レコードが耐久性を持つようになったり、故障したりすると、Zerobus Ingestはバックグラウンドでコールバックを呼び出すため、プロデューサーの進捗を遅くせずに進捗を追跡し、指標を更新でき、失敗が起きた瞬間に学習できます。

これは、オフセット待機やフラッシュとは異なります。これらはブロッキング呼び出しであり、コードはその場で永続化が完了するのを待ちます。 コールバックはブロッキングコールではありません。 これは、SDKが確認応答を届けたときに呼び出すハンドラーです。

確認コールバックはJSONおよびプロトコルバッファ(protobuf)SDKストリームでサポートされています。 Arrow Flightストリームはコールバックをサポートしていません。Arrowストリームで永続性を確認するには、wait_for_offset() または flush() を使用してください。 Zerobus Ingest で Arrow Flight を使用するを参照してください。

以下のメソッド名と型名はPython SDKからのものです。 他のZerobus SDKは、サポートされている範囲で認識コールバックを公開し、各言語で同等の構成要素を用いています。

コールバックの仕組み

コールバックは、 AckCallback をサブクラス化し、2つのメソッドを実装することで定義します。

  • on_ack(offset: int):サーバーが提出(レコードまたはバッチ)を耐久性として正常に認識したときに呼び出します。 offsetは承認された提出物を示します。
  • on_error(offset: int, error_message: str): 投稿中にエラーが発生した場合に呼ばれます。 on_error はオプションです。 故障を処理またはログ付けするために実装してください。

コールバックは、提出されたレコードやバッチの論理オフセットが認識または失敗した際に1回ずつ呼び出されるため、ストリーム全体でインジェスションの進行状況を示す継続的なシグナルとなります。

コールバックメソッドはSDKのバックグラウンドスレッド上で動作しているので、それらを呼び出してもプロデューサーがブロックされることはありません。 それらを高速かつノンブロッキングに保ちましょう。 失敗に対してどうすべきかはクライアントの責任です:記録、警告、再試行、または停止。 一部のエラーは致命的で、 on_error がストリームが永久に失敗したと報告した場合、新しいストリームで復旧しなければなりません。 回復 と再挑戦のパターンを参照してください。

コールバックの設定

ストリームにコールバックを付けるのは、ストリームを作成する際にAckCallbackサブクラスのインスタンスをStreamConfigurationOptionsack_callbackオプションとして渡すことです。 コールバックはそのストリームに取り込まれたすべてのレコードに適用されます。

from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import AckCallback, StreamConfigurationOptions, TableProperties

class MyAckCallback(AckCallback):
    def on_ack(self, offset: int) -> None:
        print(f"Record acknowledged at offset: {offset}")

    def on_error(self, offset: int, error_message: str) -> None:
        print(f"Error at offset {offset}: {error_message}")

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

table_properties = TableProperties("main.default.air_quality")
options = StreamConfigurationOptions(
    ack_callback=MyAckCallback(),
)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties, options)

try:
    for row in records:
        stream.ingest_record_offset(row)
finally:
    stream.close()

コールバックが登録されていれば、列に並ぶ必要はありません。 on_ack 各レコードが耐久性が確認された時点で発火し、 on_error が故障すると発火します。

コールバックとブロッキングの違い

コールバックとブロッキングコールは異なる問題を解決し、組み合わせて使うことができます。

  • 確認応答コールバックを使い、持続性の確認やエラーが発生した際に非同期的に反応しつつ、高いスループットを維持しましょう。 進捗追跡、指標、エラーログに適しています。
  • 特定のレコード、または保留中のすべてのレコードが先に進む前に永続化される必要がある場合は、wait_for_offset() または flush() を使用します。
  • Last updated on