流れ込むデータを受け止める — Structured Streaming と Auto Loader のファイル検出

6.1Structured Streaming 詳論 — outputMode 3種と集約の制約、withWatermark、checkpointLocation、Trigger 4種

バッチのコードなら、今日の分を読んで書けば終わります。困るのはデータが止まらないときです。5 分おきにファイルが増える入力へ同じ SELECT を毎回流すと、前回どこまで読んだかを覚えていない限り同じ行を何度も書き込みます。かといって読んだファイル名を別表に記録する仕組みを自作すると、ジョブが落ちた瞬間の整合性まで自分で面倒を見ることになります。Structured Streaming は、この「どこまで読んだか」の記憶をフレームワーク側が持つ仕組みです。記憶を預けた見返りとして、書き手はバッチとほぼ同じ DataFrame の書き方に戻れます。

readStream と writeStream — バッチとの違いはここだけ

バッチで `spark.read` と書いていたところを `spark.readStream` に、`write` を `writeStream` に替えます。間に挟む `select` や `filter` や `groupBy` はバッチと同じ式がそのまま通ります。次のコードは、クラウドストレージ上の JSON を読んで Bronze テーブルへ書き込む最小形です。

(spark.readStream
   .format("cloudFiles")
   .option("cloudFiles.format", "json")
   .option("cloudFiles.schemaLocation", "/Volumes/main/ops/chk/_schema")
   .load("/Volumes/main/raw/events")
   .writeStream
   .option("checkpointLocation", "/Volumes/main/ops/chk/bronze")
   .trigger(availableNow=True)
   .toTable("main.bronze.events"))

この処理はマイクロバッチという小さな単位に区切って動きます。到着したデータを 1 件ずつ流すのではなく、ある区間ぶんをまとめて 1 回のバッチとして実行し、それを繰り返します。頭の中では「入力が追記され続ける表があり、その表に対するクエリの結果テーブルが毎バッチ更新される」と考えると、次に出てくる outputMode の話がつながります。

ここで初学者が最もよく踏む地雷を先に潰します。ソースの読み取りに `spark.read` と書くと、増分処理になりません。`spark.read` は「その瞬間のスナップショットを 1 回読む」静的な読み取りで、何度実行しても新着を検出しません。増分で読みたいテーブルは `spark.readStream.table("main.bronze.events")` のように読みます。「新しい行が下流に流れてこない」の原因は、まずこの一語の差を疑います。

SQL しか書かないチームでも同じことができます。ソースを `STREAM()` で包み、ストリーミングテーブルとして宣言します。Databricks は、クラウドストレージからの増分取り込みで `COPY INTO` のより拡張性の高い代替として、この `CREATE STREAMING TABLE` と `read_files` の組み合わせを挙げています。

CREATE OR REFRESH STREAMING TABLE main.bronze.events
AS SELECT * FROM STREAM read_files(
  '/Volumes/main/raw/events',
  format => 'json');

もうひとつ、ストリーミングのソースは原則として追記のみ (append-only) である必要があります。上流のテーブルで既存行が更新されたり削除されたりすると、ストリームは既定で「読んだはずの行が変わった」と判断して失敗します。上流に更新が入る前提なら、変更を行単位のイベントとして扱う Change Data Feed など別の仕組みを噛ませます。

outputMode 3 種 — 何を「書き出すべき行」と呼ぶか

結果テーブルが毎バッチ更新されるとして、シンクへ実際に流すのはそのうちどれかを決めます。ここを間違えると、集約したはずのストリームが 1 行も出力されないまま静かに動き続けます。指定は `writeStream` に `.outputMode("update")` のように書きます。

この 3 つがすべてです。`incremental` や `overwrite` というモードは存在しないので、選択肢に並んでいたらその時点で誤りだと判断できます。

難しいのは、クエリの形によって使えるモードが変わる点です。集約を含まない素通しのクエリでは `append` と `update` が使え、`complete` は使えません。集約しない結果テーブルは行が無限に積み上がるため、全体を毎回書き出す意味がないからです。

逆に集約を含むクエリでは、ウォーターマークを付けていない限り `append` が使えません。集約値は後から来たデータで変わりうるので、「もう変化しない行」を宣言できないためです。この状態で `append` を指定すると、実行前の解析段階で AnalysisException になります。

# 起動前の解析でエラーになる書き方
(spark.readStream.table("main.bronze.events")
   .groupBy("device_id").count()
   .writeStream
   .outputMode("append")
   .option("checkpointLocation", "/Volumes/main/ops/chk/agg")
   .toTable("main.silver.device_counts"))

これを通すには、`withWatermark` を足してイベント時刻ウィンドウの集約にするか、`outputMode` を `update` か `complete` に変えるかの二択です。`complete` は毎バッチ全件を書き直すため、グループ数が増えるほど書き込み I/O が膨らみます。

さらにシンク側にも対応表があります。「クエリが許すモード」と「シンクが許すモード」の両方を満たす組み合わせしか動きません。

Delta テーブルへ「変わった行だけ」を反映したいときに `update` を直接指定できないのはこの表のせいです。定石は `foreachBatch` で、マイクロバッチを普通の DataFrame として受け取り、その中で `MERGE INTO` を書きます。`foreachBatch` の書き込みに `txnAppId` と `txnVersion` を渡しておくと、同じバッチが再実行されても Delta 側が重複と判定して無視します。

withWatermark — 遅れて来る 1 件のために state を抱え込まない

イベント時刻で 5 分ごとに集計しているとき、ネットワークの遅延で 3 時間前のイベントが今届くことがあります。取りこぼしたくないからと、すべての時間帯の集計途中値をメモリに残し続けると、状態 (state) が単調に増えていき、いずれエグゼキュータのメモリが枯渇します。`withWatermark` は「この幅を超えて遅れたデータはもう待たない」と宣言するためのものです。

from pyspark.sql.functions import window, count

(df.withWatermark("event_time", "10 minutes")
   .groupBy(window("event_time", "5 minutes"), "device_id")
   .agg(count("*").alias("events")))

第 1 引数はイベント時刻の列、第 2 引数は遅延の許容幅です。Spark はそのストリームで観測した最大のイベント時刻を追跡し、そこから許容幅だけ遡った時刻をしきい値とします。しきい値より古いウィンドウの state は破棄され、同時に「そのウィンドウはもう確定した」と見なせるため、`append` でも集約結果を出力できます。書き方の決まりが 2 つあり、`withWatermark` は集約より前に呼ぶことと、集約の `window()` に渡す列と同じイベント時刻列を指定することです。処理時刻の列を渡しても意味がありません。

許容幅は実測から決めます。短すぎると遅延データが黙って捨てられ、長すぎると state が肥大化して `append` の出力もその幅だけ遅れます。`"10 minutes"` と書けばウィンドウの確定は最短でも 10 分後になる、という取引です。

入力が複数あるとき、全体のしきい値は既定ですべての入力のウォーターマークの最小値になります。遅れている側に合わせる安全側の挙動です。設定は `spark.sql.streaming.multipleWatermarkPolicy` で、既定値は `min`、`max` にすると遅延は減りますが遅い入力のデータは落ちます。

checkpointLocation — 再起動の記憶をどこに置くか

クラスタは落ちますし、ランタイムのアップグレードでジョブは必ず止まります。止まったあとに同じデータをもう一度書き込まないための記憶が、`checkpointLocation` に指定したクラウド上の永続パスです。ドライバのローカルディスク (`/tmp/...`) は、クラスタが消えた時点で記憶ごと消えるため使えません。

そしてこれは省略できません。Databricks Runtime で Delta テーブルやファイルへ書く `writeStream` は、`checkpointLocation` を指定せずに起動すると `AnalysisException: checkpointLocation must be specified` でジョブ起動時に失敗します。「一時ディレクトリに保存されて再起動後も継続できる」という挙動は起きませんし、Delta の `_delta_log` が代わりを務めることもありません。`display()` や memory シンクなど一時的な場所が自動生成される例外はありますが、Databricks は耐障害性のため常に明示指定することを推奨しています。

チェックポイントに残るものは 4 種類です。各マイクロバッチで処理したソースの位置 (`offsets`)、シンクへコミット済みのマイクロバッチの記録 (`commits`)、ステートフル処理の状態 (`state`)、そしてクエリを識別する一意のクエリ ID (`metadata`) です。shuffle パーティション数などの設定はオフセットログの一部として保存されます。冪等なライターである Delta と組み合わせることで、これが exactly-once の土台になります。

運用上の決まりごとは 3 つです。第一に、クエリごとに別のチェックポイントパスを使います。複数のクエリで共有すると互いのオフセットを上書きし合って復旧できなくなります。第二に、チェックポイント内のファイルを削除するか別のパスに変えると、次回の実行は最初からやり直しになります。第三に、再起動をまたいで変えてよい設定と、変えてはいけない設定があります。フィルタの追加や削除、レート制限の値、トリガー間隔の変更は許されます。集約や join や重複排除といったステートフル処理の変更、ソースの数や種類の変更、出力スキーマの変更は許されません。

Trigger 4 種 — 次のバッチをいつ始めるか

トリガーを指定しないままにすると、常時稼働のクラスタを 24 時間押さえたまま、1 日数回しか来ないファイルを待ち続けることになります。ここはレイテンシとコストを直接交換する設定です。

実務でよく効くのは `availableNow` です。ファイルが 1 日数回しか届かないなら、ジョブのスケジュールから起動し、読み切ったらクラスタごと止めるのが最も安上がりです。似た指定の `trigger(once=True)` は Databricks Runtime 11.3 LTS 以降で非推奨です。`once` は利用可能なデータを 1 つのマイクロバッチで処理しようとするためレート制限を無視し、バックログが大きいとメモリ不足を起こします。`once` は 1 バッチ、`availableNow` は複数バッチ、と覚えます。

選び方は要件から決まります。秒単位の鮮度が必要なら既定トリガー、分単位でよくクラスタの空回りを減らしたいなら `processingTime`、到着が疎で「溜まった分をまとめて片付ける」だけでよいなら `availableNow` です。

なお `continuous` の代替として Databricks が案内しているのはリアルタイムモードで、`.trigger(realTime="5 minutes")` と書きます。この "5 minutes" はレイテンシ目標ではなくチェックポイント間隔で、Databricks Runtime 16.4 LTS 以上の classic コンピュートと `update` 出力モードが前提になります。

6.2Auto Loader (1) — ディレクトリ一覧と通知の2モード、クラウド通知リソース、状態管理と運用オプション群

ここまでで「どこまで読んだか」を Structured Streaming が覚えることは分かりました。しかし相手がクラウドストレージのフォルダになると、もうひとつ厄介な問いが残ります。そもそも新しいファイルが届いたことを、どうやって知るのかという問いです。素朴に考えれば毎回フォルダの中身を一覧すればよいのですが、1 つのプレフィックス配下に 1 億個のファイルが並ぶ本番バケットで毎分それをやると、LIST API の課金だけで請求が跳ね、一覧が返る前に次のマイクロバッチの時刻が来ます。Auto Loader は、この「検出」を担当する Structured Streaming のソースです。

使い方は `format("cloudFiles")` を指定するだけで、あとは通常のストリームと変わりません。紛らわしいのは `format` が 2 つ出てくることです。外側の `.format("cloudFiles")` は「Auto Loader を使う」という宣言で、実際のファイル形式は `.option("cloudFiles.format", "json")` のほうに書きます。`cloudFiles.format` は必須で、既定値も自動判別もありません。指定できる値は `avro` / `binaryFile` / `csv` / `json` / `orc` / `parquet` / `text` / `xml` です。画像や PDF を丸ごと取り込みたいときは `binaryFile` を選びます。

検出モードは 2 つ — 一覧するか、通知を待つか

Auto Loader のファイル検出には、ディレクトリ一覧 (directory listing) モードとファイル通知 (file notification) モードの 2 つがあります。既定はディレクトリ一覧モードで、入力パスを LIST API で走査して未処理のファイルを見つけます。追加のクラウド設定が一切要らないので、開発中や中小規模のフォルダではこれで十分です。

ファイル通知モードは、クラウドストレージ側が発行する「オブジェクトが作成された」というイベントをキュー経由で購読します。ファイルの総数が何個あろうと届いたイベントの数しか処理しないため、検出コストがディレクトリ内のファイル数に依存しません。数十万ファイルを超えたあたりから、ディレクトリ一覧モードの LIST コストと検出レイテンシは線形に悪化します。つまりモード選択の判断軸はストレージ種別でもスキーマ進化の頻度でもなく、ディレクトリ内のファイル数とファイル発見コストです。

運用中に切り替えられるかもよく問われます。検出モードはストリームの再起動をまたいで切り替えられ、切り替えても exactly-once の処理保証は維持されます。チェックポイントを消す必要も、ターゲット表を作り直す必要もありません。

クラウド側に何が作られるのか

ファイル通知モードを従来方式で使う場合は `.option("cloudFiles.useNotifications", "true")` を指定します。すると Auto Loader が、通知を受け取るためのリソースをクラウド上に自動作成します。監査で「知らないキューが増えている」と指摘されがちなので、何が作られるかは覚えておく価値があります。

Lambda や CloudWatch、EventBridge、Kinesis は使いません。選択肢にこれらが並んでいたら誤りです。すでにあるキューを使いたい場合は `cloudFiles.queueUrl` で指し示します。自動作成させるなら AWS では `s3:PutBucketNotification`、`sns:CreateTopic`、`sqs:CreateQueue` などを含む IAM ポリシーが要り、Azure では Event Grid とキューを作る Contributor ロールに加えてメッセージ操作のための Storage Queue Data Contributor が要ります。この権限付与で詰まる現場は多く、次節のファイルイベントはまさにここを回避する機能です。

通知モードには構造的な弱点もあります。クラウド事業者はファイルイベントの 100% 配信を保証していません。データの欠損が許されない要件なら、`cloudFiles.backfillInterval` に `"1 day"` のような間隔を設定して、非同期のバックフィル (取りこぼし拾い) を定期的に走らせます。並列度を上げる `cloudFiles.fetchParallelism` やバッチサイズの調整では、この取りこぼしは埋まりません。

状態管理 — 二度取り込まないことを誰が保証しているか

Auto Loader が同じファイルを二度処理しないのは、チェックポイント内にスケーラブルなキーバリューストア (RocksDB) を持ち、検出したファイルの情報をそこに永続化しているからです。MERGE で後から重複を消しているわけでも、ソースからファイルを消しているわけでもありません。この状態と Delta の冪等な書き込みが組み合わさって exactly-once になります。

この追跡には期限を設けられます。`cloudFiles.maxFileAge` は重複排除のためにファイルイベントを追跡し続ける期間で、既定値はありません。短くしすぎると追跡から外れた古いファイルが「新規」として再検出され、重複取り込みになります。指定できる最小値は `"14 days"`、Databricks の推奨は 90 日程度の余裕を持った設定です。チェックポイントのディレクトリにクラウド側のライフサイクルポリシーを当てるのも禁物で、状態が壊れて最初からやり直しになります。

状態は SQL から覗けます。`cloud_files_state` はチェックポイントのパス、またはストリーミングテーブル名を引数に取り、検出済みファイルの一覧を返す関数です。Databricks Runtime 11.3 LTS 以降で使えます。

SELECT * FROM cloud_files_state('/Volumes/main/ops/chk/bronze');

返る列には `path`、`size`、`discovery_time`、`commit_time`、`ingestion_state` などがあります。`commit_time` が NULL のファイルは、検出済みだがまだ処理されていないという意味です。「取り込まれていないファイルがある」という問い合わせは、まずこの列を見れば切り分けられます。

バックログの大きさは、ストリーミングクエリの進捗情報から分かります。Auto Loader は毎バッチ `numFilesOutstanding` (未処理ファイル数) と `numBytesOutstanding` (未処理バイト数) を報告します。ファイル通知モードでは、Databricks Runtime 10.4 LTS 以降でクラウドキュー内のイベント数の概算値 `approximateQueueSize` も得られます。スループット指標である `numInputRows` や `processedRowsPerSecond` とは役割が違い、遅れの量を見るならこの 3 つを追います。

運用オプション群 — 既定値を知らないと事故になるもの

Auto Loader のオプションは数十個ありますが、既定値を知らないまま本番に出すと事故になるものは限られています。

(spark.readStream.format("cloudFiles")
   .option("cloudFiles.format", "json")
   .option("cloudFiles.useNotifications", "true")
   .option("cloudFiles.schemaLocation", "/Volumes/main/ops/chk/_schema")
   .option("cloudFiles.inferColumnTypes", "true")
   .option("cloudFiles.maxFilesPerTrigger", "500")
   .option("cloudFiles.backfillInterval", "1 day")
   .option("pathGlobFilter", "*.json")
   .load("s3://raw-events/clickstream/"))

`maxFilesPerTrigger` と `maxBytesPerTrigger` は同時に指定できます。両方書いてもエラーにはならず、どちらか一方の上限に達するまでのファイルを処理します。`cleanSource` に `MOVE` を選んだときは `cloudFiles.cleanSource.moveDestination` で退避先を指定しますが、退避元と退避先は同じバケットまたはコンテナ内である必要があります。なお `ignoreCorruptFiles` は `cloudFiles` 名前空間のオプションではなく、Spark 共通のファイル読み取りオプションです。`cloudFiles.ignoreCorruptFiles` と書くと、`validateOptions` が既定 true なので起動時に弾かれます。

スキーマまわりも既定値が挙動を決めます。`cloudFiles.schemaLocation` に推論結果を永続化する場所を与えると、Auto Loader は最初に検出した 50 GB または 1000 ファイルのうち先に達したほうまでをサンプリングして型を決めます。この上限は `spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes` と `.numFiles` で変更できます。`schemaLocation` は読み取り側、`checkpointLocation` は書き込み側のオプションで、別物です。JSON・CSV・XML のような型情報を持たない形式では、`inferColumnTypes` が既定 false のため、入れ子のフィールドも含めて全列が文字列として推論されます。

`cloudFiles.schemaEvolutionMode` の既定は、スキーマを明示していないときが `addNewColumns`、明示しているときが `none` です。`addNewColumns` は新しい列を見つけるとスキーマへ追加したうえで `UnknownFieldException` でストリームを止め、再起動で処理を再開します。ジョブの自動リトライと組み合わせれば人手を介さずにスキーマ進化を吸収できます。型の拡張 (`int` から `long` など) まで許したい場合は `addNewColumnsWithTypeWidening` を選びます。`rescue` はスキーマを一切進化させず落ちもしないモード、`failOnNewColumns` はスキーマを手で直すまで再起動できないモード、`none` は新しい列を無視するモードです。

推論に任せきりにせず要所だけ型を固定したいときは `cloudFiles.schemaHints` を使います。書式は JSON ではなく SQL の DDL 表記で、`"user_id STRING, age INT"` のようにカンマ区切りで並べます。`"user_info.dob DATE"` のように入れ子のフィールドも指定できます。ヒントを与えた列だけがその型に固定され、指定していない列は通常どおり推論されます。

取り込む対象を絞る手段も覚えておきます。`pathGlobFilter` にグロブパターンを渡すと、`"*.png"` のようにファイル名で対象を限定できます。ディレクトリ構造から列を起こしたいときは `cloudFiles.partitionColumns` に Hive 形式のパーティション列をカンマ区切りで指定します。`cloudFiles.format` に `png` のような値は存在しないので、画像を読むなら形式は `binaryFile`、絞り込みは `pathGlobFilter`、という役割分担になります。

どのモードでも、収まらなかった値は既定名 `_rescued_data` の列に JSON 文字列として退避されます。列名は `cloudFiles.rescuedDataColumn` で変えられます。想定外の列や型不一致の値を黙って捨てるのではなく 1 列にまとめて残すため、ストリームを走らせたまま `SELECT * FROM main.bronze.events WHERE _rescued_data IS NOT NULL` で中身を検査し、送信元の修正が必要かスキーマ側で受け入れるかを判断できます。取り込みを止めずにデータ損失もゼロにする、という要件はこの列と `schemaEvolutionMode` の組み合わせで満たします。

6.3Unity Catalog 配下で取り込むときの制約 — READ FILES 権限、チェックポイントの置き場所、外部ロケーションのファイルイベント

手元の検証ワークスペースで動いていた Auto Loader が、Unity Catalog を有効にした本番で権限エラーになって止まる、という詰まり方が定番です。原因はコードではなく、読み書きする場所に対する権限と、置いてはいけない場所にチェックポイントを置いていることの 2 つに集約されます。

READ FILES — テーブルの権限では足りない

前提として、Unity Catalog がクラウドストレージを扱う構造を押さえます。まず IAM ロールやマネージド ID といったクラウド側の認証手段をストレージ資格情報として登録し、次にその資格情報とパスを組み合わせた外部ロケーションを作ります。1 つの資格情報は複数の外部ロケーションから参照できます。`dbfs:/mnt/...` のマウントは Unity Catalog のガバナンス対象外で監査も効かないため、新規に作るパイプラインでは使いません。

クラウドストレージ上のパスへの操作は、この外部ロケーションに対する専用の権限で守られます。テーブルに対する `SELECT` や `MODIFY` を持っていても、パスを直接読むことはできません。Auto Loader の実行に必要な権限は、役割ごとに分かれます。

• ソースデータのある場所 — その外部ロケーションに対する `READ FILES`
• チェックポイントとスキーマファイルを置く場所 — `READ FILES` と `WRITE FILES`、加えて `CREATE TABLE`
• 書き込み先のカタログとスキーマ — 親カタログの `USE CATALOG` と親スキーマの `USE SCHEMA`

ここでメタストア管理者やクラスタの CAN MANAGE を配ってしまうと動くには動きますが、最小権限の原則から外れます。正解は「ソースの外部ロケーションに `READ FILES` を付ける」です。もう 1 点、アクセスは2 層構造です。Unity Catalog 側の `GRANT READ FILES ON EXTERNAL LOCATION ...` に加えて、その外部ロケーションが参照するストレージ資格情報が、クラウド側でそのパスへの読み取り権限を持っている必要があります。片方だけでは通りません。

チェックポイントはテーブルの下に置けない

もうひとつの定番が配置の制約です。Unity Catalog は、チェックポイントやスキーマ推論・進化用のファイルをテーブルのディレクトリ配下に入れ子で置くことを許可していません。「ターゲット表のフォルダの下に `_checkpoint` を作る」という、Hive Metastore 時代によくやっていた整理の仕方がそのまま失敗します。DBFS ルート直下でなければならないという制約はありませんが、「置き場所に制約はない」も誤りです。

.option("checkpointLocation", "s3://dev-bucket/_checkpoint/bronze_events")
.option("cloudFiles.schemaLocation", "s3://dev-bucket/_checkpoint/bronze_events/_schema")

置き場所としては Unity Catalog ボリューム (`/Volumes/カタログ/スキーマ/ボリューム/...`) も選べます。ボリュームはテーブル以外のファイルをカタログ配下で管理する仕組みで、テーブルと同じように `GRANT` で権限を制御でき、リネージや監査ログも効きます。クエリごとに別パスという原則はここでも生きるので、テーブル名でサブディレクトリを切ります。

アクセスモードの制約は緩和されています。Databricks Runtime 11.3 LTS 以降では、標準 (旧 shared) と専用 (旧 single user) のどちらのアクセスモードでも Auto Loader を実行できます。

外部ロケーションのファイルイベント

通知モードのために SNS や SQS を作る権限を、ストリームを書く開発者ごとに配って回るのは現実的ではありません。これを解決するのが、外部ロケーションに対するファイルイベントです。外部ロケーション単位でイベントの受け口を 1 つ用意し、Databricks 側が管理します。

有効化できるのは、その外部ロケーションの所有者か `MANAGE` 権限を持つ人です。Databricks は新しく作られた外部ロケーションについては既定でファイルイベントを有効にしています。これが有効な外部ロケーションを読む Auto Loader ストリームは、通知モードを構成するために追加の権限を与える必要がありません。コード側では `cloudFiles.useManagedFileEvents` を `true` にします。Databricks Runtime 18.1 以降は既定が `if_available` で、使える場合は自動的にファイルイベントが使われます。チェックポイントが不要になるわけでも、スキーマ推論のサンプリング上限が消えるわけでもない点は取り違えないでください。

運用上の注意が 1 つあります。ファイルイベントを使う場合は、ストリームを少なくとも 7 日に 1 回は実行します。間隔が空くと保存された読み取り位置が無効になり、フルのディレクトリ一覧が必要になるためです。

この章では取り込みの入口までを扱いました。チェックポイントを使った障害復旧の手順と exactly-once の詳細は「Databricks チェックポイント」の記事、ウォーターマークと状態ストアのチューニングは「Structured Streaming」の記事、`cloudFiles` の実装の細部は「Auto Loader」の記事で深掘りします。