Structured Streaming完全ガイド｜Databricksリアルタイム処理入門

Structured Streamingは、Apache Sparkのストリーム処理フレームワークです。バッチ処理と同じDataFrame APIでリアルタイムデータを処理でき、exactly-onceのデータ処理保証を提供します。Databricksでは、Auto Loader・Delta Live Tables・イベントドリブンパイプラインの基盤として使われており、Data Engineer Associate / Professionalの両試験で頻出です。

この記事では、readStream/writeStreamの基本構文、3つの出力モード、トリガーモードの選び方、ウォーターマークによる遅延データ制御、チェックポイントの設計、foreachBatchパターンまで、コード例付きで網羅的に解説します。

基本概念：無限に追加されるテーブル

Structured Streamingでは、ストリームデータを「無限に行が追加されるテーブル（Unbounded Table）」として扱います。Kafkaトピックやクラウドストレージに新しいデータが到着するたびに、このテーブルに行が追加され、クエリの結果が増分更新されます。

開発者はバッチ処理とほぼ同じDataFrame API（select、filter、groupBy、joinなど）でストリームクエリを記述でき、Sparkエンジンが内部的にマイクロバッチ処理として実行します。

# バッチ読み込み
batch_df = spark.read.format("delta").load("/data/events")

# ストリーム読み込み（readStream に変えるだけ）
stream_df = spark.readStream.format("delta").load("/data/events")

# どちらも同じ変換を適用できる
result = stream_df.filter("event_type = 'purchase'") \
    .groupBy("region") \
    .count()

バッチとストリームの違いは入口（read vs readStream）と出口（write vs writeStream）だけです。中間の変換ロジックは共通化できるため、バッチで検証してからストリームに切り替える開発フローが実務の定番です。

readStream と writeStream の基本構文

ストリーム処理のパイプラインは、ソース読み込み（readStream）→ 変換 → シンク書き込み(writeStream)の3ステップで構成されます。

# 1. ソースからの読み込み（Auto Loader例）
raw = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoint/schema")
    .load("/data/landing/")
)

# 2. 変換
from pyspark.sql.functions import current_timestamp, col
cleaned = (raw
    .filter(col("amount") > 0)
    .withColumn("processed_at", current_timestamp())
)

# 3. シンクへの書き込み
query = (cleaned.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoint/bronze")
    .trigger(availableNow=True)
    .toTable("bronze.events")
)

主なソースとシンク

3つの出力モード（Output Mode）

outputMode は、結果テーブルのどの部分をシンクに書き出すかを制御します。クエリの種類によって使えるモードが異なるため、試験では「このクエリにこのモードは使えるか？」という形式で問われます。

appendモードと集約クエリの関係は試験で最もよく問われるポイントです。集約クエリでappendモードを使うには、必ずwithWatermark()を設定する必要があります。ウォーターマークにより結果が確定したタイミングでのみ行が出力されるためです。

completeモードは非集約クエリ（単純なfilterやselectのみ）では使えません。結果テーブル全体を保持する必要があるため、集約がないと無限にデータが蓄積されてしまうからです。

トリガーモードの比較

トリガーは「どのタイミングでマイクロバッチを実行するか」を決定します。コスト・レイテンシ・運用パターンに直結するため、実務でも試験でも頻出です。

# 実務で最も一般的: availableNow（Workflowsで定期実行）
query = (df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoint/bronze")
    .trigger(availableNow=True)
    .toTable("bronze.events")
)

# ニアリアルタイムが必要な場合: processingTime
query = (df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoint/bronze")
    .trigger(processingTime="10 seconds")
    .toTable("bronze.events")
)

availableNow=True は、Databricks Workflowsで1時間ごとに起動し、到着済みファイルを全て処理して停止するパターンで使います。常時稼働クラスタが不要なためコスト効率が最も高く、Databricksが推奨するデフォルトのトリガーです。

once=True との決定的な違いは「バッチ分割」です。onceは全データを1バッチに詰め込むため、データ量が大きいとOOMの原因になります。availableNowはmaxFilesPerTrigger等に基づいて複数バッチに分割するため、メモリ効率が高く安定します。

ウォーターマーク（withWatermark）

ウォーターマークは、イベントタイムベースのウィンドウ集約で「遅延データをどこまで待つか」の閾値を定義する仕組みです。Sparkは受信した全イベントの最大イベントタイムを追跡し、そこから閾値を引いた時刻を「ウォーターマーク」として設定します。ウォーターマークより古いイベントは無視され、対応する状態はメモリから破棄されます。

from pyspark.sql.functions import window, col

# 10分以上遅延したイベントは処理対象外
result = (stream_df
    .withWatermark("event_time", "10 minutes")
    .groupBy(
        window("event_time", "5 minutes"),
        "region"
    )
    .agg({"amount": "sum", "*": "count"})
)

# append モードで書き込む場合、ウォーターマークが必須
query = (result.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoint/windowed_agg")
    .trigger(processingTime="30 seconds")
    .toTable("gold.regional_sales")
)

ウォーターマークの動作例

• 最大イベントタイムが 12:30 でウォーターマーク閾値が10分 → ウォーターマーク = 12:20
• 12:20 より古いイベント（例: 12:15）が到着しても無視される
• 12:20 以前のウィンドウ（例: 12:10-12:15）の状態がメモリから解放される

閾値を大きくするほど遅延データを拾えますが、保持する状態量が増えてメモリ消費が上がります。逆に小さすぎると正当なデータを取りこぼします。実務では、データの遅延特性を計測してから閾値を決定します。

チェックポイント

チェックポイントは、ストリームクエリの進捗状態（処理済みオフセット・集約の中間結果・メタデータ）をクラウドストレージに永続化する仕組みです。障害発生時にチェックポイントから再開することで、exactly-once処理を保証します。

# checkpointLocation はwriteStreamに必須
query = (df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "dbfs:/checkpoint/pipeline_bronze")
    .trigger(availableNow=True)
    .toTable("bronze.events")
)

チェックポイントの設計ルール

Auto Loaderとの組み合わせ

Auto Loader（cloudFilesフォーマット）は、Structured Streamingの上に構築されたファイル取り込み機能です。クラウドストレージに到着したファイルを自動検知し、増分処理でDelta Tableに取り込みます。readStreamのフォーマットに"cloudFiles"を指定するだけで利用できます。

# Auto Loader + Structured Streaming の典型パターン
raw = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoint/schema")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load("s3://bucket/landing/events/")
)

# Bronze Tableへの書き込み
query = (raw.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoint/bronze_events")
    .option("mergeSchema", "true")
    .trigger(availableNow=True)
    .toTable("bronze.raw_events")
)

Auto Loaderのチェックポイントには「どのファイルを処理済みか」が記録されるため、同じファイルが二重に処理されることはありません。スキーマ推論結果はcloudFiles.schemaLocationに保存され、再起動時にはこの保存済みスキーマが使われます。

foreachBatch パターン

foreachBatchは、各マイクロバッチの結果をカスタム関数で処理する仕組みです。Delta TableへのMERGE（upsert）、複数テーブルへの同時書き込み、外部APIの呼び出しなど、標準のwriteStreamでは実現できない処理に使います。

from delta.tables import DeltaTable

def upsert_to_silver(batch_df, batch_id):
    """マイクロバッチごとにMERGE（upsert）を実行"""
    silver = DeltaTable.forName(spark, "silver.customers")

    (silver.alias("t")
        .merge(batch_df.alias("s"), "t.customer_id = s.customer_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )

# foreachBatch でMERGEを実行
query = (df.writeStream
    .foreachBatch(upsert_to_silver)
    .option("checkpointLocation", "/checkpoint/silver_customers")
    .trigger(availableNow=True)
    .start()
)

foreachBatch の典型ユースケース

• MERGEによるupsert: ストリームデータで既存レコードを更新しつつ新規レコードを挿入。CDC（Change Data Capture）パイプラインで定番
• 複数テーブルへの書き込み: 1つのマイクロバッチを複数のDeltaテーブルやKafkaトピックに同時出力
• 外部システム連携: バッチ単位でREST APIに送信したり、データベースにJDBC書き込みしたりする

foreachBatch内では通常のDataFrame操作がすべて使えるため、batch_df.createOrReplaceTempView()でSparkSQL経由のMERGEも可能です。ただし、foreachBatch内でエラーが発生するとそのバッチ全体がリトライされる点に注意が必要です。冪等な処理を書くか、MERGEのように重複適用しても結果が変わらない操作を使うのが鉄則です。

試験で問われるポイント

• readStream / writeStream の構文: バッチとの違いは入口・出口だけ。中間の変換ロジックは共通
• 出力モードの制約: 集約クエリでappendモード → ウォーターマーク必須。非集約でcompleteモード → 使用不可
• availableNow vs once: availableNowは複数バッチに分割（推奨）、onceは単一バッチ（非推奨）
• ウォーターマーク: 遅延データの許容閾値を定義。閾値超過のデータは無視され、状態がメモリから解放される
• チェックポイント: exactly-once保証の基盤。1クエリ1ディレクトリ。削除すると全件再処理
• foreachBatch: MERGEや複数シンクへの書き込みに使う。冪等性の確保が必須
• Auto Loader連携: cloudFilesフォーマットはStructured Streamingの上に構築されている

サンプル問題

よくある質問