回復および再試行パターン

この詳細解説では、Lakeflow Connect で回復力のある Zerobus Ingest クライアントを構築する方法を説明します。これには、SDK の組み込みリカバリ、エラーの表面化の仕組み、およびストリームが完全に失敗したときに未確認のレコードを救済する方法が含まれます。以下のメソッド名とオプション名は、Python SDK のものです。他の SDK でも同等の機能が公開されています。

組み込みリカバリ

Zerobus SDK は、一時的なエラーから自動的に回復します。ストリームが再試行可能なエラー（通常はタイムアウトやネットワーク切断）に遭遇したとき、またはストリームが正常なシャットダウン信号を受信したときに、リカバリが Trigger されます。リカバリはdefaultでオンになっており、ストリーム構成オプションを通じて調整できます：

オプション	説明
recovery	自動ストリーム復旧を有効にします。
recovery_timeout_ms	復旧操作のタイムアウト。
recovery_backoff_ms	リカバリ試行間の遅延。
recovery_retries	リカバリの最大試行回数。

オプション	説明
recovery	自動ストリーム復旧を有効にします。
recovery_timeout_ms	復旧操作のタイムアウト。
recovery_backoff_ms	リカバリ試行間の遅延。
recovery_retries	リカバリの最大試行回数。

ほとんどのワークロードでは、defaultが出発点として適しており、一時的な問題に対して独自の再接続ループを作成する必要はありません。SDK を使用する場合、OAuth トークンもストリームの作成時およびリカバリ時に自動的に更新されるため、クライアント側で管理する必要はありません。例外は REST API であり、この場合はクライアント自身が OAuth トークンの取得と更新を行います。Zerobus Ingest の使用を参照してください。これらのオプションのdefault値と単位については、Zerobus SDKリポジトリを参照してください。

エラーと再試行

SDK は、ネットワークの問題や一時的なサーバーエラーなどの一時的なエラーを、組み込みの回復機能を通じて自動的に再試行します。無効な認証情報やテーブルの欠落など、回復できない障害は ZerobusException として表面化します。ZerobusException をキャッチして障害に対処し、根本的な原因を修正するか、新しいストリームで回復するか、停止するかを決定します。

Python

from zerobus.sdk.shared import ZerobusException

try:
    stream.ingest_record_offset(record)
except ZerobusException as e:
    # Handle the failure: log it, fix the cause, recover on a new stream, or stop.
    ...

エラーコードの完全なリストについては、Zerobus Ingest エラー処理を参照してください。

未確認レコードのリカバリ

ストリームが永続的に失敗し、SDK の自動回復が使い果たされた後でも、送信済みでサーバーによってまだ確認されていないレコードは、Asynchronous communication で説明されているインフライトバッファにクライアントによって保持されます。データを失わないように、それらを取得してください：

• get_unacked_records() 承諾されていないレコードを未加工のバイトとして返します。
• get_unacked_batches() バッチ再試行ロジックのために、承諾されていないバッチ（それぞれレコードのリスト）を返します。

レコードはシリアル化された形式で返されます。json.loads(record.decode('utf-8')) を使用して JSON をデコードするか、メッセージ型を使用してプロトコルのバッファ (protobuf) を逆シリアル化します。それらを保存するか、新しいストリームでリプレイします。

永続的な障害後のストリームの回復

SDK は、一時的なエラーに対する再試行を自動的に処理します。エンキュー、フラッシュ、およびクローズの失敗はすべて ZerobusException として表面化します。get_unacked_records() および recreate_stream() は、ストリームがすでに閉じられた後にのみ成功します。これは致命的なエラーが発生した場合の状態です。エンキューに失敗してもストリームはアクティブなままとなるため、それらの呼び出しは失敗します。その場合は、元のエラーを発生させ、ストリームを保持してください。recreate_stream() は、すでに承諾されたレコードを再キューイングします。エンキューに失敗したペイロードは再試行されません。

Python

from zerobus.sdk.shared import ZerobusException

try:
    for i in range(10000):
        stream.ingest_record_offset(record)
    stream.flush()
except ZerobusException as e:
    print(f"Ingestion failed: {e}")
    try:
        unacked = list(stream.get_unacked_records())
    except ZerobusException:
        raise e
    print(f"{len(unacked)} previously queued records were unacknowledged.")
    try:
        new_stream = sdk.recreate_stream(stream)
        try:
            new_stream.flush()
        finally:
            new_stream.close()
    except ZerobusException:
        raise e
else:
    stream.close()

ストリームが閉じた後に元のバッチグループを調査するには、get_unacked_batches() を使用します：

Python

unacked_batches = list(stream.get_unacked_batches())
print(f"{len(unacked_batches)} batches remain unacknowledged")

リプレイ時の重複の処理

Zerobus Ingestは「最低1回」の配信を提供しますが、「厳密に1回」ではないため、救済されたレコードの再生や再試行によって、レコードが複数回書き込まれる可能性があります。ワークロードで重複が許容されない場合は、レイクハウス内のデータを重複排除してください：

• 各レコードに安定した一意の識別子（たとえば、ソースが割り当てたイベント ID やナチュラルキーなど）を含めます。
• 読み取り時または下流の処理中に重複排除を行います。例えば、識別子に一致する MERGE INTO や、変換処理におけるウィンドウ化された ROW_NUMBER() を使用します。

ストリーム上のレコードは順序どおりにコミットされるため、単調増加するシーケンス番号は重複排除キーとしても有効です。

フラッシュと正常終了

• flush() ストリームを閉じずに、送信したレコードが耐久性のあるものとしてサーバーによって認識されるのを待ちます。ストリームの途中で耐久性のチェックポイントが必要な場合に呼び出します。
• close() ストリームを正常にフラッシュして閉じ、保留中のレコードが永続的であると確認されるのを待ってから戻ります。ストリームの失敗からの回復ではなく、正常なシャットダウンに使用してください。ストリームが失敗した場合は、「永続的な失敗後のストリームの回復」に示すように、代わりに未確認のレコードを救済してください。

ストリーム全体ではなく特定のレコードの耐久性を確認するには、メッセージのブロックと確認 を参照してください。

永続的なfallbackロケーションからのデータ復旧

Zerobus Ingest がデータを永続化した後、公開される前にターゲットテーブルに破壊的変更が加えられた場合、Zerobus Ingest はそのデータを破棄するのではなく、テーブルのストレージルート下の fallback ディレクトリに Parquet ファイルとして書き込みます。Durable fallback location を参照してください。

データがそこに書き込まれたことを確認する方法： fallbackディレクトリは _zerobus/table_rejected_parquets/ であり、テーブルの物理的なルートストレージロケーションからの相対パスです。取り込みが継続しているにもかかわらず、テーブル変更後にテーブルから行が欠落している場合は、そのディレクトリでParquetファイルを確認してください。

fallbackデータをテーブルに再処理する

原因を修正したら（通常はテーブルスキーマをプロデューサーが送信するものと一致させることで修正します）、fallbackのParquetファイルをターゲットテーブルに再処理してください。ファイルはテーブルのストレージ場所にある標準的なParquetファイルであるため、COPY INTOを使用してロードできます：

1. スキーマの不一致を解決します。 fallbackレコードが適合するように、ターゲットテーブル（またはプロデューサーのスキーマ）を進化させます。スキーマ管理を参照してください。

2. ロードする前にfallbackデータを検査してください。 fallbackパスに対してクエリーを実行し、そこに何があるか、そしてそれがテーブルと一致していることを確認します：

   SQL

   SELECT * FROM parquet.`<table-storage-root>/_zerobus/table_rejected_parquets/` LIMIT 10;

3. COPY INTO を使用してファイルをロードします。 COPY INTO は冪等です。すでにロードされたファイルを追跡するため、再実行しても同じ fallback ファイルが二重にロードされることはありません：

   SQL

   COPY INTO <catalog>.<schema>.<table>
   FROM '<table-storage-root>/_zerobus/table_rejected_parquets/'
   FILEFORMAT = PARQUET
   COPY_OPTIONS ('mergeSchema' = 'false');

4. 期待される行数が書き込まれたことを 検証 します。データがテーブルにあることを確認したら、fallbackディレクトリが不要になった場合にクリーンアップしてください。

Zerobus Ingest は at-least-once（最低1回）であるため、テーブルに公開され、かつ fallback ロケーションに書き込まれたレコードは、2 回ロードされる可能性があります。重複が問題となる場合は、リプレイ時の重複の処理で説明されているように重複排除を行ってください。継続的または自動化された再処理を行うには、COPY INTO を手動で実行する代わりに、Auto Loader を fallback パスに向けることができます。

注記

この手順は、fallback の Parquet ファイルに対して標準の Delta ツールを使用する一般的な初回アプローチです。本番運用で利用する前に、テーブルおよびストレージ設定に対して検証を行ってください。

関連

• メッセージのブロックと確認応答：取り込み方法と耐久性の確認。
• Zerobus Ingest のエラー処理: 完全なエラーコードリファレンス。
• Durable fallback location：The durable fallback location。