Databricks Structured Streamingのチェックポイント運用: 再開性・Exactly-once・障害復旧

Structured Streamingのチェックポイントは、ストリーミングクエリの進捗状態を永続化し、障害時の再開とExactly-once処理を実現する基盤です。 設計を誤るとデータの重複・欠損・ステート肥大化によるOOMが発生します。 本稿ではチェックポイントの内部構造、設計ルール、障害復旧手順、Exactly-onceの達成条件を実務に即して解説します。

チェックポイントの内部構造

チェックポイントディレクトリには4つのサブディレクトリが生成されます。 それぞれがストリーミングクエリの異なる状態を管理しています。

/mnt/checkpoints/orders-pipeline/
├── offsets/
│   ├── 0    # バッチ0: {"kafka-topic": {"0": 100, "1": 150}}
│   ├── 1    # バッチ1: {"kafka-topic": {"0": 200, "1": 300}}
│   └── 2
├── commits/
│   ├── 0    # バッチ0のシンク書き込み完了
│   └── 1    # バッチ1のシンク書き込み完了
├── sources/
│   └── 0/
└── state/
    └── 0/
        └── delta/  # ステートストアのスナップショット

再起動時のリカバリロジックは次の通りです。offsetsに記録があるがcommitsに対応するバッチIDがない場合、そのバッチは「読み取り済み・未コミット」と判断され、再処理されます。 commitsに記録がある場合はスキップされるため、二重書き込みが防止されます。

設計ルール: 1クエリ1チェックポイント

チェックポイントの最重要ルールは「1つのストリーミングクエリに1つの固有のチェックポイントパス」を割り当てることです。

• 複数クエリで同一パスを共有すると、offsetsファイルが競合してデータ欠損・重複が発生する
• 同一クエリでもロジック変更時に古いチェックポイントを再利用すると、ステートの不整合でエラーになる場合がある
• パス命名は /mnt/checkpoints/{pipeline-name}/{query-name}/ のような階層構造を推奨

# クエリごとに固有のcheckpointLocationを指定
orders_query = (
    orders_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/orders-to-silver")
    .toTable("silver.orders")
)

payments_query = (
    payments_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/payments-to-silver")
    .toTable("silver.payments")
)

チェックポイントの保存先選定

本番環境ではクラウドストレージまたはUC Volumesを使用し、DBFS上のチェックポイントは開発・検証用に限定します。

Exactly-once処理の達成条件

Structured StreamingのExactly-onceセマンティクスは、ソースのリプレイ可能性とシンクの冪等性の両方が揃って初めて成立します。

Deltaテーブルをシンクにする場合、DeltaのトランザクションログとStructured Streamingのチェックポイントが協調してExactly-onceを実現します。 foreachBatchで外部システムに書き込む場合は、バッチIDを使った冪等キーの管理が必要です。

def process_batch(batch_df, batch_id):
    # バッチIDを使った冪等処理の例
    batch_df.write \
        .format("delta") \
        .mode("append") \
        .saveAsTable("silver.events")

(source_df.writeStream
    .foreachBatch(process_batch)
    .option("checkpointLocation", "/mnt/checkpoints/events-pipeline/events-to-silver")
    .start()
)

障害復旧手順

ストリーミングジョブが障害で停止した場合の復旧パターンを整理します。

パターン1: 単純な再起動（最も一般的）

同じcheckpointLocationを指定してジョブを再起動します。チェックポイントから最終コミット済みバッチの次から再開されます。 コードのロジックに変更がない場合はこれで十分です。

パターン2: ロジック変更を伴う復旧

• 出力カラムの追加・削除: mergeSchemaで対応可能な場合はチェックポイント継続可
• ステートフル操作の変更（ウォーターマーク期間変更、集約キー変更）: チェックポイントをリセットし、シンクテーブルも再構築
• ソースの変更（トピック追加等）: チェックポイントをリセットして新規開始

パターン3: チェックポイント破損時の復旧

# 1. 破損したチェックポイントを退避
dbutils.fs.mv(
    "/mnt/checkpoints/orders-pipeline/orders-to-silver",
    "/mnt/checkpoints/_archived/orders-to-silver-corrupted-20260327"
)

# 2. シンクテーブルの最終タイムスタンプを確認
last_ts = spark.sql("""
    SELECT MAX(event_timestamp) FROM silver.orders
""").collect()[0][0]

# 3. 新チェックポイントで再開（重複防止にMERGEを使用）
(source_df.writeStream
    .foreachBatch(lambda df, id: merge_deduplicate(df, id))
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/orders-to-silver")
    .start()
)

ステート管理とウォーターマーク

ウィンドウ集約やstream-streamジョインでは、ステートがチェックポイントのstate/ディレクトリに蓄積されます。 ウォーターマークを設定しないとステートが無限に肥大化し、OOMを引き起こします。

from pyspark.sql.functions import window, col

(orders_df
    .withWatermark("event_time", "10 minutes")
    .groupBy(window("event_time", "5 minutes"), "product_id")
    .agg({"amount": "sum"})
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/windowed-agg")
    .toTable("gold.orders_5min_agg")
)

• ウォーターマークは「この時間より古いデータはもう来ない」という宣言
• ウォーターマーク期限を過ぎたステートは自動的にパージされ、メモリとチェックポイントサイズを抑制する
• 期限が短すぎると遅延データが破棄され、長すぎるとステートが肥大化するトレードオフ

運用チェックリスト

• checkpointLocationはクラウドストレージまたはUC Volumesに配置する（DBFSは開発のみ）
• 1クエリ1チェックポイントパスの命名規則を統一する
• ウォーターマークをすべてのステートフル操作に設定する
• チェックポイントディレクトリのサイズを定期監視し、異常な肥大化を検出する
• 障害復旧手順をRunbookに文書化し、チェックポイントリセット時のシンクテーブル処理も明記する
• ジョブの再試行ポリシーで自動復旧を設定する（DatabricksジョブのRetries設定）

問題で確認

よくある質問