Professional 学習教材 ⑦ データインジェスト（配点 7%）

本教材は Databricks 認定データエンジニア Professional（上位資格） 試験のドメイン⑦「データインジェスト」の自習用教材です。Professional 試験では Auto Loader に加え、Structured Streaming による取り込み（チェックポイント、trigger、foreachBatch、ウォーターマーク、ステートフル処理、冪等シンク）が深く問われます。このファイル単体で学習が完結するよう、公式ドキュメントの定義・用語・仕組み・オプション・注意点を網羅しています。
参照した公式ページ:
https://learn.microsoft.com/ja-jp/azure/databricks/ingestion/ （Lakeflow Connect 標準コネクタ / インジェストの 3 レイヤー）
https://learn.microsoft.com/ja-jp/azure/databricks/ingestion/cloud-object-storage/auto-loader/ （Auto Loader とは）
https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/ （構造化ストリーミング概要）
https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/concepts （構造化ストリーミングの概念）
https://learn.microsoft.com/ja-jp/azure/databricks/structured-streaming/checkpoints （チェックポイント）
補足参照（同ドキュメント群、用語網羅のため実読）:
.../structured-streaming/triggers （トリガー間隔）
.../structured-streaming/output-mode （出力モード）
.../structured-streaming/foreach （foreachBatch / foreach）
.../structured-streaming/watermarks （ウォーターマーク）
.../structured-streaming/delta-lake （Delta Lake ストリーミング読み書き）
.../ingestion/cloud-object-storage/auto-loader/schema （スキーマ推論・進化）
.../ingestion/cloud-object-storage/auto-loader/production （運用ワークロード構成）
.../ingestion/cloud-object-storage/copy-into/ （COPY INTO）

1. このドメインの概要

データインジェスト（Data Ingestion） とは、外部のクラウドオブジェクトストレージ、メッセージバス、データベース、SaaS アプリケーションなどから、Databricks の Lakehouse（Delta Lake テーブル）へデータを取り込む工程を指します。Professional 試験では単にツール名を覚えるだけでなく、「どの方式をどの要件で選ぶか」「exactly-once（厳密に 1 回）をどう担保するか」「チェックポイント・トリガー・出力モードの相互作用」を理解しているかが問われます。

Databricks のインジェストは、カスタマイズ性と自動化度のトレードオフで 3 つのレイヤー に整理されます（詳細は 3-1）。

Structured Streaming（構造化ストリーミング） — 最もカスタマイズ性が高い。Spark API で exactly-once のエンドツーエンド・フォールトトレランスを提供するストリーミングエンジン。
Lakeflow パイプライン（宣言的パイプライン、旧 DLT） — Structured Streaming を拡張した宣言的フレームワーク。オーケストレーション・監視・データ品質・エラー処理を自動管理。
マネージドコネクタ（Lakeflow Connect） — Salesforce・SQL Server など特定ソース向けのフルマネージド。CDC・認証・自動再試行・自動スキーマ進化まで内包。

Databricks は「最も管理されたレイヤーから始め、要件を満たせない場合に下位（よりカスタマイズ可能な）レイヤーに降りる」ことを推奨します。

このドメインの中核は次の 2 つです。

Auto Loader（cloudFiles ソース）: クラウドオブジェクトストレージからのファイルの増分取り込みのデファクト。
Structured Streaming: Auto Loader の基盤でもあり、Kafka などメッセージバスや Delta テーブル間ストリーミングにも使う汎用増分処理エンジン。

インジェストのスケジュール は 2 モードに大別されます。

利用シーン	パイプラインモード	対応するトリガー
バッチインジェスト	トリガー（Triggered）: スケジュール実行または手動実行時に新しいデータを処理	`Trigger.AvailableNow`
ストリーミングインジェスト	継続（Continuous）: ソースに到着した新しいデータを継続処理	既定 / `processingTime`

2. 重要用語集

用語（日本語）	English	説明
データインジェスト	Data Ingestion	外部ソースから Lakehouse へデータを取り込む工程。バッチとストリーミングがある。
自動ローダー	Auto Loader	クラウドストレージに到着した新規ファイルを増分・効率的に処理する仕組み。`cloudFiles` ソースを提供。
cloudFiles	cloudFiles	Auto Loader が提供する Structured Streaming ソース名。`spark.readStream.format("cloudFiles")` で使用。
増分検知 / ファイル検出モード	Incremental detection / File detection mode	新規ファイルを見つける仕組み。ディレクトリ一覧モードとファイル通知モードの 2 種類。
ディレクトリ一覧モード	Directory listing mode	ストレージの `LIST` API でディレクトリを列挙し新規ファイルを検出。既定モード。設定不要だがファイル数が多いと高コスト。
ファイル通知モード	File notification mode	クラウドの通知サービス・キューを使い、ディレクトリ列挙を回避して低コスト・低レイテンシで検出。Databricks が多くのワークロードで推奨。
ファイルイベント	File events	ファイル通知モードの新方式。バケットごとに 1 キューで増分検出。
スキーマ推論	Schema inference	読み込みデータのスキーマを自動検出する機能。先頭 50GB または 1000 ファイルをサンプリング。
スキーマ進化	Schema evolution	新しい列を検出した際にテーブルスキーマを自動的に発展させる機能。`cloudFiles.schemaEvolutionMode` で制御。
スキーマの場所	Schema location	`cloudFiles.schemaLocation`。スキーマ情報を時系列で追跡・保存するディレクトリ。
スキーマヒント	Schema hints	`cloudFiles.schemaHints`。推論スキーマに対し既知の型情報を明示指定してオーバーライド。
レスキューされたデータ列	Rescued data column	`_rescued_data`。スキーマに一致しない（欠落・型不一致・大文字小文字不一致）データを削除せず JSON BLOB として保存する列。
構造化ストリーミング	Structured Streaming	Spark のリアルタイムに近い処理エンジン。バッチと同じ API で増分処理し exactly-once を保証。
無制限テーブル	Unbounded table	ストリームを、行が継続的に追記され続ける（終端のない）入力テーブルとして扱う概念モデル。
readStream	readStream	ストリーミングソースを読み取る API。`spark.readStream.format(...)...load()`。
writeStream	writeStream	ストリーミングシンクへ書き込む API。`df.writeStream.option("checkpointLocation", ...).start()`。
ソース	Source	ストリームの入力（Auto Loader、Delta テーブル、Kafka、Pub/Sub、Pulsar など）。
シンク	Sink	ストリームの出力先（Delta テーブル、Kafka、foreach、memory、console など）。
チェックポイント	Checkpoint	処理状態・進行状況を保存し、フォールトトレランスと exactly-once を実現する場所。`checkpointLocation` で指定。
先書きログ / WAL	Write-Ahead Log (WAL)	チェックポイントと連携し、処理を保証する仕組み。オフセットログ・コミットログを含む。
オフセット	Offset	各マイクロバッチで処理したソースの位置。再処理せず中断地点から再開するために記録。
コミット	Commit	どのマイクロバッチがシンクにコミットされたかの記録。exactly-once の要。
トリガー	Trigger	新しいデータをチェックする頻度・方式を制御する設定。
Trigger.ProcessingTime	Trigger.ProcessingTime	固定間隔マイクロバッチ。`.trigger(processingTime='10 seconds')`。コストとレイテンシのバランス。
Trigger.AvailableNow	Trigger.AvailableNow	起動時点で利用可能な全データを増分バッチで処理して停止。`.trigger(availableNow=True)`。スケジュール実行に推奨。
Trigger.Once	Trigger.Once	使用可能な全データを 1 回で処理。Databricks Runtime 11.3 LTS 以降で非推奨。`AvailableNow` を使う。
Trigger.Continuous	Trigger.Continuous	Spark OSS の実験的連続処理。Databricks ではサポート・非推奨外。代わりにリアルタイムモード。
リアルタイムモード	Real-time mode	超低レイテンシ（1 秒未満、代表 ~300ms）モード。`.trigger(realTime='5 minutes')`（引数はマイクロバッチ長）。パブリックプレビュー。
出力モード	Output mode	各トリガーで演算子が出力するレコード種別。append / update / complete。
追加モード	Append mode	既定。将来変更されない行のみ出力。ステートフルではウォーターマーク経過後に確定出力。
更新モード	Update mode	トリガー中に変更された全行を出力（後で再変更されうるものも含む）。
完全モード	Complete mode	ストリーミング集計のみ。状態テーブルの全結果行を毎回出力。データ増大で性能低下。
foreachBatch	foreachBatch	各マイクロバッチ出力 DataFrame にバッチ関数を適用。MERGE / 複数シンク / 既存バッチライター再利用に使用。
foreach	foreach	行単位のカスタム書き込み。連続処理モードで動く（foreachBatch は不可）。
厳密に 1 回	Exactly-once	障害があっても各レコードが結果に 1 回だけ反映されるセマンティクス。チェックポイント + 冪等シンクで実現。
少なくとも 1 回	At-least-once	重複の可能性がある配信保証。`foreachBatch` 自体はこれのみ保証。
冪等シンク / 冪等書き込み	Idempotent sink / write	同じバッチを再実行しても結果が重複しない書き込み。Delta では `txnAppId` + `txnVersion`。
ウォーターマーク	Watermark	ステートフル操作で遅延データを待つ時間の閾値。古い状態を削除しメモリを制御。`withWatermark(col, delay)`。
ステートフル処理	Stateful processing	集計・ストリーム間結合・重複除去など中間状態を保持する処理。
ステートレス処理	Stateless processing	状態を保持せず行を処理する処理。全出力モードで同一挙動。
重複除去	Deduplication	`dropDuplicates()` / `dropDuplicatesWithinWatermark()` / `distinct()` による重複排除。
ウィンドウ集計	Window aggregation	タンブリング / スライディング / セッションウィンドウでの時間ベース集計。
ストリーム間結合	Stream-stream join	2 つのストリームの結合。追加モードのみ、外部結合はウォーターマーク必須。
read_files	read_files	SQL のテーブル値関数。`STREAM read_files(...)` で Auto Loader をストリーミングテーブルに使用。
STREAMING TABLE	Streaming Table	`CREATE STREAMING TABLE`。SQL でスケーラブルな増分取り込みを行う推奨構文（`COPY INTO` より推奨）。
COPY INTO	COPY INTO	SQL コマンド。ファイルを Delta テーブルに冪等・増分ロード。読込済ファイルはスキップ。
Lakeflow Connect	Lakeflow Connect	Databricks のインジェスト製品群（標準コネクタ + マネージドコネクタ）。
Lakeflow パイプライン	Lakeflow Pipelines (旧 DLT)	Structured Streaming を拡張した宣言的 ETL フレームワーク。スキーマ/チェックポイントを自動管理。
マネージドコネクタ	Managed connectors	Lakeflow パイプライン上に構築されたフルマネージドのソース別コネクタ。CDC・自動スキーマ進化を内包。
RocksDB	RocksDB	Auto Loader が検出済みファイルのメタデータを保持するチェックポイント内のキー/値ストア。
バックフィル	Backfill	既存の（過去の）ファイルをまとめて取り込むこと。`cloudFiles.backfillInterval` で定期実行も可能。

3. 詳細解説

3-1. インジェストの全体像（3 レイヤー：Streaming / 宣言的パイプライン / マネージドコネクタ）

Databricks の ETL スタックは、最もカスタマイズ可能なもの → 最も管理されたもの の順に 3 レイヤーで構成されます。一部のコネクタは単一レイヤー（例：Salesforce・SQL Server のフルマネージドコネクタ）で、他は複数レイヤーにまたがります（例：クラウドオブジェクトストレージからの取り込みは 3 レイヤーすべてで選べる）。

レイヤー	説明	特徴
構造化ストリーミング	Spark Structured Streaming。Spark API で exactly-once のエンドツーエンド・フォールトトレランスを提供するストリーミングエンジン。	最もカスタマイズ性が高い。スキーマ・チェックポイントを自分で管理。Python / Scala。
Lakeflow パイプライン	構造化ストリーミングを拡張した宣言的フレームワーク。変換を定義するとオーケストレーション・監視・データ品質・エラー処理を管理。	自動化が多くオーバーヘッドが少ない。Python / SQL。スキーマ・チェックポイントを自動管理。
マネージドコネクタ	Lakeflow パイプライン上に構築されたフルマネージド。ソース固有の認証・CDC・エッジケース処理・API 保守・自動再試行・自動スキーマ進化を追加。	最も管理された体験。サポート対象ソースに最大の自動化。

選択の原則: 最も管理されたレイヤーから始め、要件（例：ソースが未サポート）を満たせなければ下位レイヤーへドロップダウンする。

ソース別の代表的な選択肢（カスタマイズ大 ← → 自動化大）:

ソース	カスタマイズ大	一部カスタマイズ	自動化大
クラウドオブジェクトストレージ	構造化ストリーミング + Auto Loader (Python/Scala)	Lakeflow パイプライン + Auto Loader (Python/SQL)	Databricks SQL + Auto Loader (SQL)
Apache Kafka	構造化ストリーミング + Kafka (Python/Scala)	Lakeflow パイプライン + Kafka (Python/SQL)	Databricks SQL (SQL)
Google Pub/Sub, Apache Pulsar	構造化ストリーミング (Python/Scala)	Lakeflow パイプライン (Python/SQL)	Databricks SQL (SQL)
SFTP サーバー	SFTP から取り込み (Python/SQL)	N/A	N/A

SQL でのファイル取り込み: クラウドオブジェクトストレージからの増分取り込みを SQL で行う場合、COPY INTO ではなく CREATE STREAMING TABLE（内部で read_files = Auto Loader） が、スケーラブルで堅牢なため推奨されます。

COPY INTO vs. ストリーミングテーブル/Auto Loader（3-2 末尾と 4 章も参照）:

COPY INTO: SQL コマンドで冪等・増分ロード。読込済ファイルは後続実行でスキップ。シンプルな一括・定期ロード向き。ただし大量ファイルではスケーラビリティに限界。
STREAMING TABLE / Auto Loader: 数十億ファイル・毎時数百万ファイルまでスケール。SQL ユーザーにも推奨。

3-2. Auto Loader の仕組み（増分検知・スキーマ推論/進化・チェックポイント・exactly-once）

概要

Auto Loader は、クラウドストレージに到着した新規データファイルを 追加設定なしで増分・効率的に処理 します。cloudFiles という Structured Streaming ソースを提供し、入力ディレクトリを指定すると新規ファイルを自動処理（既存ファイルの処理も可能）。移行時の数十億ファイル処理やテーブルのバックフィルにも使え、毎時数百万ファイル のほぼリアルタイム取り込みまでスケールします。

サポートソース（ストレージ）

Amazon S3（s3://）
Azure Data Lake Storage / ADLS（abfss://）
Google Cloud Storage / GCS（gs://）
Unity Catalog ボリューム（/Volumes/）
Azure Blob Storage（wasbs://、レガシ WASB は非推奨。ABFS 推奨）

サポートファイル形式

JSON, CSV, XML, PARQUET, AVRO, ORC, TEXT, BINARYFILE。事前圧縮ファイルの読み取りも可。

増分検知（ファイル検出モード）

Auto Loader は 2 つの検出モードをサポートします。ファイルの検出・処理の順序は保証されません（どちらのモードでも）。

モード	仕組み	コスト・特徴
ディレクトリ一覧モード（既定）	ネイティブクラウド API の `LIST` でディレクトリを列挙し新規ファイルを検出。設定不要。	ファイル数が多いと `LIST` コスト増。継続トリガーだと特に高コスト。
ファイル通知モード	ストレージのファイル通知サービス/キューを自動設定し、ディレクトリ列挙を回避。	Databricks が多くのワークロードで推奨。検出コストを大幅削減。低レイテンシ。

ファイルソースへの直接 Structured Streaming（spark.readStream.format(fileFormat).load(dir)）より Auto Loader が優れる点:

スケーラビリティ: 数十億ファイルを効率検出。バックフィルを非同期実行可能。
パフォーマンス: 検出コストはディレクトリ数ではなく 取り込むファイル数 に比例。
スキーマ推論と進化のサポート: スキーマドリフトを検出し通知。無視・喪失を防止。
コスト: ネイティブクラウド API で列挙。ファイル通知モードならディレクトリ列挙自体を回避。

進行状況の追跡と exactly-once

ファイルが検出されると、そのメタデータはチェックポイントの場所にある スケーラブルなキー/値ストア（RocksDB） に保持されます。このストアにより データは厳密に 1 回だけ処理 されます。エラー時はチェックポイントの情報から中断地点で再開し、Delta Lake への書き込みでも exactly-once を維持します。フォールトトレランスや exactly-once のために 状態を自分で維持・管理する必要はありません。

スキーマ推論の仕組み

初回読み取り時、最初に検出した 50GB または 1000 ファイル（先に達した方）をサンプリングして推論。
スキーマ情報は cloudFiles.schemaLocation ディレクトリ内の _schemas に保存し、時系列で変化を追跡。
サンプルサイズは SQL 構成で変更可能:
- spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes（例 10gb）
- spark.databricks.cloudFiles.schemaInference.sampleSize.numFiles（整数）
既定の推論型（型不一致による進化問題を回避するため保守的）:

形式	既定推論型
JSON / CSV / XML	すべて文字列（JSON の入れ子フィールド含む）
Avro	Avro スキーマでエンコードされた型
Parquet	Parquet スキーマでエンコードされた型

JSON/CSV/XML でも列の型を実データから推論させたい場合は cloudFiles.inferColumnTypes=true。
CSV はヘッダー有りと仮定。無い場合 .option("header","false")。
2 つの Parquet 間で型が違う場合は 最も広い型 を選択（schemaHints で上書き可）。

レスキューされたデータ列（`_rescued_data`）

スキーマ推論時に自動追加される列。スキーマに一致しない列を削除せず保存 します。次の理由で解析されなかったデータが JSON BLOB（+ソースファイルパス）として入ります。

スキーマに列が無い
型が一致しない
大文字小文字が一致しない

rescuedDataColumn オプションで列名変更・明示追加が可能。_rescued_data があれば、型不一致でも DROPMALFORMED（削除）や FAILFAST（エラー）にはならず、破損レコード（不完全/不正な JSON・CSV）のみが失敗扱いになります。

スキーマ進化の仕組みとモード

Auto Loader は処理中に新規列を検出すると、まず最新マイクロバッチにスキーマ推論を行い、スキーマの保存場所を 新列を末尾に追加 して更新し、UnknownFieldException でストリームを停止 します（既存列の型は変わらない）。Databricks はスキーマ変更後に自動再起動するよう Lakeflow ジョブでストリームを構成することを推奨。

cloudFiles.schemaEvolutionMode のモード:

モード	新列を読んだ時の動作
`addNewColumns`（スキーマ未指定時の既定）	新列をスキーマに追加後、`UnknownFieldException` で失敗。再起動で更新スキーマで処理再開。既存列の型は進化しない。
`addNewColumnsWithTypeWidening`	`addNewColumns` と同じ + サポート型の拡大（`int`→`long` 等）。非対応の型変更（`int`→`string` 等）はレスキュー列へ。DBR 16.4+ でパブリックプレビュー。
`rescue`	スキーマは進化させず、失敗もしない。新列はすべてレスキュー列に記録。
`failOnNewColumns`	スキーマ更新または問題ファイル削除まで失敗し再起動しない。自動更新なし。
`none`	進化せず新列を無視。`rescuedDataColumn` 未設定ならデータも復旧しない。失敗もしない。

注意: スキーマを指定した場合の既定は none。addNewColumns はスキーマ指定時は不可（ただしスキーマヒントとして指定した場合は機能）。

スキーマヒント

cloudFiles.schemaHints で推論スキーマに既知の型情報を適用（例: "tags map<string,string>, version int"）。user_info.dob DATE のような入れ子指定、配列・マップ要素指定（products ARRAY<INT>, ids MAP<STRING,INT> 等）も可。ストリームに存在しない列の追加も可能。スキーマを指定しない場合にのみ 使われ、inferColumnTypes の有無に関係なく使用可。

パーティション

Hive スタイル（base_path/event=click/date=2021-04-01/f0.json）なら event, date をパーティション列として推論。競合や非 Hive 構造なら無視。進化ではパーティション列は考慮されないため、新パーティション列を捕捉するには cloudFiles.partitionColumns を明示（例 event,date,hour）。

大文字小文字

大文字小文字を区別しない限り abc/Abc/ABC は同一列扱い（サンプルから任意選択）。区別させるにはスキーマヒントや設定。レスキュー有効時、非選択ケースの列は _rescued_data に入る。readerCaseSensitive=false で区別しない読み取り。

3-3. Structured Streaming の基礎（ソース/シンク・出力モード・トリガー・チェックポイント/WAL）

基本モデル

Structured Streaming は リアルタイムに近い処理エンジン で、バッチと同じ Spark API を使い、エンドツーエンドのフォールトトレランス と 厳密に 1 回だけの処理 を保証します。静的データへのバッチ計算と同じようにストリーミングデータへの計算を記述でき、エンジンはデータ受信と並行して段階的に計算し、結果を継続更新します（無制限テーブルに継続的に行が追記されるモデル）。

読み取り（ソース）

spark.readStream で増分取り込み。主なソース:

Auto Loader（cloudFiles）: クラウドストレージの新規ファイル。
Delta Lake テーブル: exactly-once のストリーミングソース/シンク。
標準コネクタ: メッセージバス・キュー・エンタープライズアプリ（Kafka / Pub/Sub / Pulsar など）。
マイクロバッチサイズ制御: 入力レート制限で一貫したバッチサイズを維持し遅延を防止。

書き込み（シンク）と配信構成

df.writeStream でターゲットに配信。構成要素:

チェックポイント: 処理状態を保存し、フォールトトレランスと exactly-once を実現。
出力モード: append / update / complete（ステートフル集計で構成が必要）。
トリガー間隔: レイテンシとコストのバランス。
リアルタイムモード: 5ms 級のエンドツーエンド低レイテンシ。

代表的シンク: Delta テーブル（toTable）、Kafka、foreach/foreachBatch、memory、console。

出力モード（詳細）

出力モードは「各トリガーで演算子が出力するレコード種別」を決めます。構成が必要なのは集計を含むステートフルストリームのみ。ステートレスでは全モード同一挙動。結合は append のみ サポート、重複除去は出力モードに影響されない、mapGroupsWithState/flatMapGroupsWithState は独自ロジックで出力。

出力モード	説明
追加（append）既定	将来のトリガーで変更されない行のみ出力。ステートフルではウォーターマークで確定判定。
更新（update）	トリガー中に変更された全行を出力（後で再変更されうるものも含む）。
完全（complete）	ストリーミング集計のみ。生成された全結果行を毎回出力。データ増大で性能低下。

シンク別サポート:

Kafka: 全出力モードをサポート。
Delta Lake（Unity Catalog マネージドテーブル）: append と complete をサポート、update は非サポート。update 相当の動作は foreachBatch + MERGE（ストリーミングでのマージ）で実現。

選択指針:

ダウンストリームが「書き込みごとに 1 アクション」（通知など）→ append（各レコード 1 回のみ）。
ダウンストリームが常に最新結果を要する（ML 特徴量・リアルタイムダッシュボード）→ update。
complete はデータ増大で性能劣化するため、多くのステートフル処理では マテリアライズドビュー の利用が推奨。
レイテンシ/コスト: append はウォーターマーク繰り延べ期間だけ遅延（例 1h なら少なくとも 1h 遅れる）。update は集計値ごとにトリガー毎に書き込むため、レコード単位課金のシンクでは高コストになりうる。

トリガー（詳細）

トリガー間隔は「新しいデータをチェックする頻度」を制御します。設定しないと processingTime=0（数ミリ秒ごと） となり、大量のストレージ API 呼び出しで 予期しないクラウド課金 が発生しうるため、必ずユースケースに合ったトリガーを設定します。

トリガーモード	構文（Python）	最適用途
未指定（既定）	N/A	3〜5 秒レイテンシの汎用ストリーミング（`processingTime=0` と同じ）。新データがある限り継続実行。
processingTime	`.trigger(processingTime='10 seconds')`	コストと性能のバランス。頻繁なチェックを抑えオーバーヘッド軽減。「固定間隔マイクロバッチ」。
AvailableNow	`.trigger(availableNow=True)`	スケジュールされた増分バッチ処理。起動時点で利用可能な全データを処理して停止。`maxBytesPerTrigger` 等でバッチサイズ調整可。
realTime	`.trigger(realTime='5 minutes')`	超低レイテンシ（1 秒未満、代表 ~300ms）。引数はマイクロバッチ長。パブリックプレビュー。
continuous	`.trigger(continuous='1 second')`	非サポート（Spark OSS の実験的機能）。代わりにリアルタイムモード。

Trigger.Once は DBR 11.3 LTS 以降で非推奨 → 増分バッチは Trigger.AvailableNow を使う。
サーバーレスコンピュート では Trigger.AvailableNow() と Trigger.Once() のみサポート（AvailableNow 推奨）。継続ストリーミングは Lakeflow パイプラインの継続モードを使う。
AvailableNow のソース別最小 DBR: ファイルソース 9.1 LTS、Delta/Auto Loader/Kafka 10.4 LTS、Kinesis 13.1 など。
トリガー間隔は同じチェックポイントで変更可能（増分バッチ⇔時間間隔）。ただしクエリエラーからの復旧はトリガー変更では解決しない（前回失敗バッチの完了が必要 = 冪等マイクロバッチ要件）。復旧はコンピュートのスケールアップで対処、まれに新チェックポイントで再起動。

チェックポイントと先書きログ（WAL）

チェックポイントと WAL が連携して処理を保証 します。チェックポイントは状態情報や処理済みレコードなどクエリを識別する情報を追跡。チェックポイントディレクトリのファイルを削除するか新しい場所に変えると、次回実行は新規開始 になります。

チェックポイントディレクトリの内容:

オフセット（offsets）: 各マイクロバッチで処理するソースオフセット。中断地点から再処理なしで再開。
コミット（commits）: どのマイクロバッチがシンクにコミットされたか。exactly-once を実現。
状態（state）: ステートフルクエリ（集計・ストリーム間結合・重複除去・transformWithState）のステートフル演算子・状態スキーマ・状態ストア内容のメタデータ。
メタデータ（metadata）: クエリを識別する一意のクエリ ID。構成設定はオフセットログの一部として保存。

重要ルール: クエリごとに異なるチェックポイントの場所が必要。複数クエリで同一の場所を共有してはいけない。

有効化（省略不可、display() や memory シンクは一時チェックポイントを自動生成するがフォールトトレランス保証は無いので明示推奨）:

python

(df.writeStream
  .option("checkpointLocation", "/Volumes/catalog/schema/volume/path")
  .toTable("catalog.schema.table")
)

再起動時に許可される変更 / 許可されない変更:

一般に安全: フィルターの追加/削除、レート制限変更、トリガー間隔変更、mapGroupsWithState 内の UDF ロジック更新（セマンティクスは変わりうる）。
新チェックポイントが必要（既定で不許可）: 入力ソースの数・種類の変更、購読 Kafka トピック / Auto Loader パスの変更、ステートフル操作の種類・状態スキーマ、出力シンクの種類。
ステートフル操作の変更は再起動間で不許可（状態スキーマが同一である必要）: ストリーミング集計のグループ化キー/集計の数・種類、重複除去キー、ストリーム間結合のスキーマ/等結合列/結合種類、任意ステートフル操作の状態スキーマ/タイムアウト種類。
シンク種類の変更は個別検証（例: ファイル→Kafka は可、Kafka→ファイルは不可、Kafka⇔foreach は可）。出力ディレクトリ変更は不可、Kafka トピック変更は可。
注意: dropDuplicates() / dropDuplicatesWithinWatermark() は コンピュートアクセスモードを跨ぐと状態スキーマ互換チェックで再起動失敗 しうる。専用⇔分離なし、標準⇔サーバーレスは可、他の組合せは変更しない。

ソースの進化（Source evolution, DBR 18.2+）: 既定ではソースはクエリプラン内の位置（0,1,2…）で識別されるため、ソースの数・順序を変えるとチェックポイント非互換。spark.sql.streaming.queryEvolution.enableSourceEvolution=true にして各ソースに .name() で安定名を付けると、チェックポイント状態を失わずにソースの並べ替え・追加・削除が可能（英数字とアンダースコアのみ、クエリ内で一意）。ソース命名には新チェックポイントが必要で、有効化は不可逆・ソース名は永続。

3-4. 高度な取り込み（foreachBatch、冪等な書き込み、ウォーターマーク、ステート）

foreachBatch

streamingDF.writeStream.foreachBatch(fn) で、各マイクロバッチの出力 DataFrame にバッチ関数を適用します。関数は 2 引数を受け取ります。

マイクロバッチ出力の DataFrame
マイクロバッチの一意の batchId

主な用途:

Delta Lake の MERGE（upsert）: ストリーミングでのマージは foreachBatch の使用が必須。集計結果を update 相当で Delta に反映。
既存バッチデータソースの再利用: Structured Streaming 非対応のシンク（Cassandra, Azure Synapse など）に既存のバッチライターで書ける。
複数の場所への書き込み: ただし foreachBatch 内で複数シンクに書くと 書き込みがシリアル化 されレイテンシ増。Databricks は シンクごとに個別の Structured Streaming ライターを使う ことを推奨。
ストリーミング DataFrame では未対応の DataFrame/Dataset 操作を、マイクロバッチ単位で適用できる。

重要な保証と注意:

foreachBatch() は「少なくとも 1 回（at-least-once）」しか保証しない。exactly-once を得るには batchId を使って出力を重複除去 する（エンドツーエンドのセマンティクスは自分で設計）。
連続処理モードでは動かない（マイクロバッチ実行に依存）。連続モードで書くなら foreach() を使う。
空の DataFrame を受け取りうる（Delta ソースの OPTIMIZE で処理対象ファイルがない、述語プッシュダウン/ファイルプルーニングで全レコード削除など）。空を処理しないとクエリが失敗しうる → if not output_df.isEmpty(): ...。
ステートフル演算子を使う場合は各バッチの DataFrame を完全に消費する。show(2) などで一部しか使わないと次バッチで失敗しうる → 残りを batch_df.foreach(do_nothing) で消費。
DBR 14.0（標準アクセスモード）での挙動変更: print() はドライバーログへ、dbutils.widgets はアクセス不可、関数内参照物はシリアライズ可能である必要。
エラー処理: Databricks は「クエリを高速に失敗させ、Lakeflow ジョブや Airflow などオーケストレーション層に再試行を任せる」ことを推奨（コード内の複雑な再試行ループはデータ損失リスク）。ロジック/スキーマエラー・致命的エラーは伝播、一時的シンクエラー（HTTP 429・タイムアウト等）はキャッチして再試行/DLQ。
デッドレターキュー（DLQ）: 不正レコードでクエリ全体を止めず、セカンダリ Delta テーブルにルーティングして有効データの処理を継続。監査/コンプライアンス要件に有用。

冪等なテーブル書き込み（Delta, `txnAppId` / `txnVersion`）

foreachBatch は at-least-once のため、Delta への冪等書き込み で exactly-once 相当を実現します。

txnAppId: DataFrame 書き込みごとに渡す一意の文字列（例: StreamingQuery ID。ストリーム ID に紐付ける必要はない）。
txnVersion: トランザクションバージョンとして機能する単調増加の数値（batchId にバインドする）。

Delta Lake は txnAppId + txnVersion で 重複書き込みを識別・無視 します。エラーで中断後、同じ txnAppId・txnVersion でバッチを再実行すれば重複がスキップされます。

python

app_id = "..."  # 一意のアプリケーション ID

def writeToDeltaLakeTableIdempotent(batch_df, batch_id):
  batch_df.write.format("delta").option("txnVersion", batch_id).option("txnAppId", app_id).save("...")  # location 1
  batch_df.write.format("delta").option("txnVersion", batch_id).option("txnAppId", app_id).save("...")  # location 2

streamingDF.writeStream.foreachBatch(writeToDeltaLakeTableIdempotent).start()

警告: チェックポイントを削除して新チェックポイントで再起動する場合は 別の txnAppId を指定 する（新チェックポイントは batchId 0 から始まり、Delta は (batchId, txnAppId) を一意キーとして既知バッチをスキップするため）。また foreachBatch 内の MERGE は冪等であること（そうでないと再起動時に同一バッチを複数回適用しうる）。merge は入力データを複数回読むため入力レートメトリックが実レートの倍数になりうる → merge 前に DataFrame をキャッシュ、後に解放。

Delta Lake ストリーミングの補足（ソース側の変更処理）

ストリーミングソースの Delta は 追加入力のみ 受け付ける。ソースで UPDATE/DELETE/MERGE INTO/OVERWRITE が起きるとストリームは失敗。対処は 4 通り: skipChangeCommits（変更を無視し追加のみ処理、新規ワークロード推奨）、完全更新、変更データフィード（CDF, 全種変更を処理・最も堅牢）、マテリアライズドビュー。
レガシオプション: ignoreDeletes（パーティション境界の削除のみ）、ignoreChanges（DBR 12.2 LTS で skipChangeCommits に置換）。
開始位置: startingVersion（Delta のバージョン、latest で最新のみ）/ startingTimestamp（両者同時指定は不可）。ストリーミングソースのスキーマは常に 最新スキーマ。
入力レート制限: maxFilesPerTrigger（既定 1000）、maxBytesPerTrigger（ソフト上限、未設定が既定）。両方指定時はどちらかの上限到達までを 1 バッチとする。
ソーステーブルの保持: ストリームは保持期間内に少なくとも 1 回実行が必要（VACUUM 既定 7 日、logRetentionDuration 30 日）。遅れると DELTA_FILE_NOT_FOUND_DETAILED で失敗し完全更新が必要。

ウォーターマーク

ステートフルクエリは時間とともに状態を蓄積します。ウォーターマークは古い状態を自動削除 し、メモリエラーや遅延増大を防ぎます。ストリーミング DataFrame にタイムスタンプ列と遅延しきい値を宣言します。

python

from pyspark.sql.functions import window
(df
  .withWatermark("event_time", "10 minutes")
  .groupBy(window("event_time", "5 minutes"), "id")
  .count()
)

状態エンティティの例: 時間ウィンドウ集計、ストリーム間結合の一意キー。
しきい値内に到着したレコードは必ず処理。しきい値超のレコードは処理される場合もあるが保証されない。
トレードオフ: 短いウォーターマーク = 低レイテンシだが遅延データ許容度が低い。長いウォーターマーク = 遅延データに強いが高レイテンシ・状態増大。
groupBy()/window() は列名・col()・（Scala）$colName で参照してイベント時間マーカーを保持すること。

ウィンドウ集計 × 出力モード:

出力モード	動作
追加	ウォーターマーク超過後に確定行を書き込み（遅延しきい値だけ遅れる）。古い集計状態を削除。
更新	結果計算のたびに書き込み・上書き。しきい値到達で古い状態を削除。
完全	状態を削除せず毎トリガーでテーブルを書き換え。

ストリーム間結合: 追加モードのみ。内部結合は各ソースにウォーターマークを推奨（無いと毎トリガーで両側の全キー結合を試み性能悪化）。外部結合はウォーターマーク必須（不一致は null 行、遅延しきい値到達まで書かれない）。

複数ウォーターマークポリシー: 各入力に withWatermark で個別しきい値を設定。既定はグローバルウォーターマークに 最小値（min）を採用し、遅いストリームのペースに合わせて誤った遅延判定を防ぐ。spark.sql.streaming.multipleWatermarkPolicy=max で最速ストリーム基準にできるが遅いストリームのデータが落ちるため慎重に。

重複除去:

distinct(): 状態内の一意レコードを追跡。ウォーターマーク無しだと状態が無限増大。
dropDuplicatesWithinWatermark(["guid"])（DBR 13.3 LTS+）: ウォーターマークしきい値内で任意フィールド（イベント時刻等が異なっても）で重複除去。ウォーターマーク指定が必須。全重複を保証するにはしきい値を重複イベント間の最大タイムスタンプ差より大きく設定。
特定列の重複除去は dropDuplicates() ではなく dropDuplicatesWithinWatermark() または distinct() を使う。

ウィンドウ種類:

タンブリング（tumbling）: 重複しない固定サイズ。各行は 1 ウィンドウ。例: 1 時間ごとの売上。
スライディング（sliding）: 重複可能な固定サイズ。1 行が複数ウィンドウに属す。slideDuration <= windowDuration。例: ローリング 6 時間売上。
セッション（session）: 可変サイズ。行到着で開き、ギャップ期間（例 30 分）無入力で閉じる。TimestampType/TimestampNTZType が必要。月以上の期間は非サポート。

ステートフル処理まとめ

ステートフル: 集計・ストリーム間結合・重複除去・任意ステートフル（mapGroupsWithState/flatMapGroupsWithState/transformWithState）。中間状態を保持し、フォールトトレラントストレージへ自動チェックポイント、再起動時に復元（状態スキーマは同一である必要）。
ステートレス: 状態を持たず、全出力モードで同一挙動。

4. 構文・コード例

4-1. Auto Loader（cloudFiles）: readStream → writeStream（スキーマ推論・進化つき）

python

(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "parquet")            # json/csv/avro/xml など
  .option("cloudFiles.schemaLocation", "<path-to-schema>")  # スキーマ推論/進化を有効化
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path-to-target>")
)

schemaLocation は checkpointLocation と同じディレクトリでもよい。
型を実データから推論: .option("cloudFiles.inferColumnTypes", "true")
スキーマ進化モード: .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
スキーマヒント: .option("cloudFiles.schemaHints", "version int, tags map<string,string>")

4-2. Auto Loader をスケジュール実行（Trigger.AvailableNow + レート制限）

python

df = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint)
  .load("s3://my-bucket/landing/")
)

(df.writeStream
  .option("checkpointLocation", checkpoint)
  .trigger(availableNow=True)                # 起動時点の全データを増分バッチ処理して停止
  .toTable("my_catalog.my_schema.raw_events")
)

レート制限: .option("cloudFiles.maxFilesPerTrigger", 1000)（既定1000, ハード上限）、.option("cloudFiles.maxBytesPerTrigger", "10g")（ソフト上限）。

4-3. SQL（Lakeflow パイプライン / STREAMING TABLE = Auto Loader）

sql

CREATE OR REFRESH STREAMING TABLE raw_events
AS SELECT * FROM STREAM read_files(
  's3://my-bucket/landing/',
  format => 'json'
);

4-4. トリガーモードの例

python

# 固定間隔マイクロバッチ（コストと性能のバランス）
df.writeStream.trigger(processingTime='10 seconds').toTable("t")

# 増分バッチ（スケジュール実行, Trigger.Once の後継）
df.writeStream.trigger(availableNow=True).toTable("t")

4-5. 出力モード

python

df.writeStream.outputMode("append").toTable("target_table")   # 既定
df.writeStream.outputMode("update").toTable("target_table")   # Delta は非サポート → foreachBatch+MERGE
df.writeStream.outputMode("complete").toTable("target_table") # 集計のみ

4-6. ウォーターマーク + ウィンドウ集計

python

from pyspark.sql.functions import window, sum
hourly_sales = (orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window("timestamp", "1 hour"))
  .agg(sum("amount").alias("total_sales"))
)

4-7. foreachBatch で冪等な Delta 書き込み（exactly-once 相当）

python

app_id = "orders-streaming-job"

def process_batch(batch_df, batch_id):
    if batch_df.isEmpty():                 # 空バッチを処理
        return
    (batch_df.write.format("delta").mode("append")
        .option("txnVersion", batch_id)    # batchId にバインド
        .option("txnAppId", app_id)        # 一意のアプリID
        .saveAsTable("catalog.schema.orders"))

(spark.readStream.format("delta").table("catalog.schema.raw_orders")
  .writeStream
  .foreachBatch(process_batch)
  .option("checkpointLocation", "/path/to/checkpoint")
  .start())

4-8. foreachBatch + MERGE（ストリーミング upsert / update 相当）

python

def upsertToDelta(microBatchOutputDF, batchId):
  microBatchOutputDF.createOrReplaceTempView("updates")
  microBatchOutputDF.sparkSession.sql("""
    MERGE INTO aggregates t
    USING updates s
    ON s.key = t.key
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)

(streamingAggregatesDF.writeStream
  .foreachBatch(upsertToDelta)
  .outputMode("update")
  .start())

4-9. 重複除去（ウォーターマーク内）

python

(streamingDf
  .withWatermark("eventTime", "10 hours")
  .dropDuplicatesWithinWatermark(["guid"])
)

4-10. COPY INTO（SQL, 冪等な一括/定期ロード）

sql

COPY INTO <catalog>.<schema>.booking_updates_upload
FROM '/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates'
FILEFORMAT = JSON
FORMAT_OPTIONS ('multiLine' = 'true');
-- 冪等: 繰り返し実行しても新規データのみロード。読込済ファイルはスキップ。

5. 試験で問われるポイント

Professional 試験で狙われやすい観点を整理します。

exactly-once の実現メカニズム
- Auto Loader: 検出ファイルのメタデータを RocksDB（チェックポイント内） に保持 → 厳密に 1 回。
- Structured Streaming: チェックポイント（オフセット + コミット）+ WAL で exactly-once。Delta シンクはトランザクションログで保証。
- foreachBatch は at-least-once のみ → batchId（txnVersion）+ txnAppId で冪等化して exactly-once 相当にする。
チェックポイントの原則
- クエリごとに別の場所が必須、共有不可。
- チェックポイント削除 or 場所変更 = 新規開始。
- 内容: offsets / commits / state / metadata（クエリ ID）。
- 再起動で変えてはいけないもの（ソース数・種類、ステートフル操作/状態スキーマ、シンク種類）と、変えてよいもの（フィルター、レート制限、トリガー間隔）。
トリガーの選択
- Trigger.Once は非推奨 → Trigger.AvailableNow（増分バッチ、スケジュール実行、コスト削減）。
- トリガー未設定は processingTime=0 で高頻度 API 呼び出し = 予期しないクラウド課金。
- サーバーレスは AvailableNow/Once のみ。
- Trigger.Continuous は Databricks 非サポート（→ リアルタイムモード）。
出力モードとシンクの互換性
- Delta は update 非サポート（append/complete のみ）→ update 相当は foreachBatch + MERGE。
- complete は集計のみ・データ増大で劣化 → マテリアライズドビュー推奨。
- append はウォーターマーク繰り延べ期間だけ遅延。結合は append のみ。
スキーマ推論・進化
- サンプリング = 50GB または 1000 ファイル（先着）。
- JSON/CSV/XML の既定推論は すべて文字列（inferColumnTypes=true で変更）。
- 新列検出時 addNewColumns（既定）は UnknownFieldException で停止 → Lakeflow ジョブで自動再起動。
- rescue は失敗しない・failOnNewColumns は失敗し再起動しない・none は無視。
- _rescued_data はスキーマ不一致（欠落/型/大小文字）データを削除せず保持。
ウォーターマークとステートフル処理
- 遅延データ待機と 古い状態削除 のトレードオフ。
- 外部ストリーム間結合はウォーターマーク必須。複数ウォーターマークは既定 min。
- dropDuplicatesWithinWatermark は DBR 13.3 LTS+、ウォーターマーク必須。
ファイル検出モードとコスト
- ディレクトリ一覧（既定、LIST コスト） vs ファイル通知（推奨、低コスト・低レイテンシ）。
- 継続トリガー × ディレクトリ一覧は高コスト → ファイル通知 or AvailableNow バッチ。
- cloudFiles.maxFileAge（最小 14 days, 推奨は 90 日程度）で状態肥大化を制御（過度な短縮は重複/欠落リスク）。
- チェックポイントの場所には クラウドライフサイクルポリシーを設定しない（削除されると状態破損）。
方式の使い分け
- 大量ファイル・スキーマ進化・低コスト増分 → Auto Loader。
- シンプルな SQL 一括/定期ロード・冪等 → COPY INTO（ただし大規模は STREAMING TABLE 推奨）。
- SQL でスケーラブル増分 → CREATE STREAMING TABLE（read_files）。
- 3 レイヤー: 最も管理されたレイヤーから始め、要件を満たせなければ下位へ。
foreachBatch の実務注意
- 空 DataFrame を処理する。ステートフル時はバッチを完全消費。
- 複数シンクへはシリアル化されるので、シンクごとに別ライター が推奨。
- 連続処理では動かない（→ foreach）。エラーはオーケストレーション層に再試行を委譲。
Delta ソース側の変更処理
- ソースの UPDATE/DELETE/MERGE/OVERWRITE はストリーム失敗 → skipChangeCommits（推奨）/ CDF / 完全更新 / マテリアライズドビュー。
- startingVersion / startingTimestamp で開始位置指定（同時指定不可、スキーマは常に最新）。

6. 理解度チェックリスト

[ ] インジェストの 3 レイヤー（構造化ストリーミング / Lakeflow パイプライン / マネージドコネクタ）を、カスタマイズ性と自動化度の順に説明できる。
[ ] 「最も管理されたレイヤーから始め、要件を満たせなければ下位に降りる」という選択原則を説明できる。
[ ] Auto Loader が cloudFiles ソースを提供し、RocksDB（チェックポイント内）で exactly-once を実現する仕組みを説明できる。
[ ] ディレクトリ一覧モードとファイル通知モードの違い・コスト特性・推奨を説明できる。
[ ] Auto Loader がファイルソースへの直接 Structured Streaming より優れる 4 点（スケーラビリティ/性能/スキーマ/コスト）を挙げられる。
[ ] スキーマ推論のサンプリング（50GB または 1000 ファイル）と、JSON/CSV/XML の既定が文字列である点を説明できる。
[ ] cloudFiles.schemaEvolutionMode の 5 モード（addNewColumns / addNewColumnsWithTypeWidening / rescue / failOnNewColumns / none）の挙動を区別できる。
[ ] _rescued_data（レスキュー列）が何を保存するか（欠落/型/大小文字不一致）を説明できる。
[ ] cloudFiles.schemaHints と cloudFiles.schemaLocation の役割を説明できる。
[ ] スキーマ進化時に UnknownFieldException で停止し、Lakeflow ジョブで自動再起動する運用を説明できる。
[ ] Structured Streaming の readStream / writeStream、ソース / シンクの対応を挙げられる。
[ ] チェックポイントディレクトリの 4 要素（offsets / commits / state / metadata）を説明できる。
[ ] 「クエリごとに別チェックポイント」「削除/変更で新規開始」というルールを説明できる。
[ ] WAL（オフセットログ・コミットログ）とチェックポイントの連携で exactly-once が成る仕組みを説明できる。
[ ] トリガー 4 種（未指定 / processingTime / AvailableNow / realTime）と continuous 非サポートを区別できる。
[ ] Trigger.Once が非推奨で Trigger.AvailableNow に置き換わったこと、サーバーレス制約を説明できる。
[ ] トリガー未設定（processingTime=0）が予期しないクラウド課金を招く理由を説明できる。
[ ] 出力モード（append / update / complete）の違いと、どのクエリ・シンクで使えるかを説明できる。
[ ] Delta シンクが update 非サポートで、foreachBatch + MERGE で代替する点を説明できる。
[ ] ウォーターマーク（withWatermark）が遅延データ待機と状態削除を制御することを説明できる。
[ ] タンブリング / スライディング / セッションウィンドウの違いを説明できる。
[ ] ストリーム間結合で外部結合がウォーターマーク必須である点、複数ウォーターマーク既定が min である点を説明できる。
[ ] dropDuplicatesWithinWatermark と dropDuplicates / distinct の使い分けを説明できる。
[ ] foreachBatch が at-least-once のみ保証で、batchId による重複除去で exactly-once 相当になることを説明できる。
[ ] 冪等な Delta 書き込みの txnAppId / txnVersion（batchId にバインド）の仕組みと、チェックポイント再作成時に別 txnAppId が必要な理由を説明できる。
[ ] foreachBatch の注意点（空 DataFrame 処理、バッチ完全消費、複数シンクのシリアル化、連続処理非対応）を挙げられる。
[ ] Delta ストリーミングソースでの skipChangeCommits / CDF / 完全更新 / マテリアライズドビューの使い分けを説明できる。
[ ] startingVersion / startingTimestamp と maxFilesPerTrigger / maxBytesPerTrigger の役割を説明できる。
[ ] COPY INTO の冪等性・増分ロードと、STREAMING TABLE（read_files）が推奨される理由を説明できる。
[ ] cloudFiles.maxFileAge（最小 14 日、推奨 90 日程度）と、チェックポイントにライフサイクルポリシーを設定しない理由を説明できる。

Professional 学習教材 ⑦ データインジェスト（配点 7%） ​

1. このドメインの概要 ​

2. 重要用語集 ​

3. 詳細解説 ​

3-1. インジェストの全体像（3 レイヤー：Streaming / 宣言的パイプライン / マネージドコネクタ） ​

3-2. Auto Loader の仕組み（増分検知・スキーマ推論/進化・チェックポイント・exactly-once） ​

概要 ​

サポートソース（ストレージ） ​

サポートファイル形式 ​

増分検知（ファイル検出モード） ​

進行状況の追跡と exactly-once ​

スキーマ推論の仕組み ​

レスキューされたデータ列（_rescued_data） ​

スキーマ進化の仕組みとモード ​

スキーマヒント ​

パーティション ​

大文字小文字 ​

3-3. Structured Streaming の基礎（ソース/シンク・出力モード・トリガー・チェックポイント/WAL） ​

基本モデル ​

読み取り（ソース） ​

書き込み（シンク）と配信構成 ​

出力モード（詳細） ​

トリガー（詳細） ​

チェックポイントと先書きログ（WAL） ​

3-4. 高度な取り込み（foreachBatch、冪等な書き込み、ウォーターマーク、ステート） ​

foreachBatch ​

冪等なテーブル書き込み（Delta, txnAppId / txnVersion） ​

Delta Lake ストリーミングの補足（ソース側の変更処理） ​

ウォーターマーク ​

ステートフル処理まとめ ​

4. 構文・コード例 ​

4-1. Auto Loader（cloudFiles）: readStream → writeStream（スキーマ推論・進化つき） ​

4-2. Auto Loader をスケジュール実行（Trigger.AvailableNow + レート制限） ​

4-3. SQL（Lakeflow パイプライン / STREAMING TABLE = Auto Loader） ​

4-4. トリガーモードの例 ​

4-5. 出力モード ​

4-6. ウォーターマーク + ウィンドウ集計 ​

4-7. foreachBatch で冪等な Delta 書き込み（exactly-once 相当） ​

4-8. foreachBatch + MERGE（ストリーミング upsert / update 相当） ​

4-9. 重複除去（ウォーターマーク内） ​

4-10. COPY INTO（SQL, 冪等な一括/定期ロード） ​

5. 試験で問われるポイント ​

6. 理解度チェックリスト ​

Professional 学習教材 ⑦ データインジェスト（配点 7%）

1. このドメインの概要

2. 重要用語集

3. 詳細解説

3-1. インジェストの全体像（3 レイヤー：Streaming / 宣言的パイプライン / マネージドコネクタ）

3-2. Auto Loader の仕組み（増分検知・スキーマ推論/進化・チェックポイント・exactly-once）

概要

サポートソース（ストレージ）

サポートファイル形式

増分検知（ファイル検出モード）

進行状況の追跡と exactly-once

スキーマ推論の仕組み

レスキューされたデータ列（`_rescued_data`）

スキーマ進化の仕組みとモード

スキーマヒント

パーティション

大文字小文字

3-3. Structured Streaming の基礎（ソース/シンク・出力モード・トリガー・チェックポイント/WAL）

基本モデル

読み取り（ソース）

書き込み（シンク）と配信構成

出力モード（詳細）

トリガー（詳細）

チェックポイントと先書きログ（WAL）

3-4. 高度な取り込み（foreachBatch、冪等な書き込み、ウォーターマーク、ステート）

foreachBatch

冪等なテーブル書き込み（Delta, `txnAppId` / `txnVersion`）

Delta Lake ストリーミングの補足（ソース側の変更処理）

ウォーターマーク

ステートフル処理まとめ

4. 構文・コード例

4-1. Auto Loader（cloudFiles）: readStream → writeStream（スキーマ推論・進化つき）

4-2. Auto Loader をスケジュール実行（Trigger.AvailableNow + レート制限）

4-3. SQL（Lakeflow パイプライン / STREAMING TABLE = Auto Loader）

4-4. トリガーモードの例

4-5. 出力モード

4-6. ウォーターマーク + ウィンドウ集計

4-7. foreachBatch で冪等な Delta 書き込み（exactly-once 相当）

4-8. foreachBatch + MERGE（ストリーミング upsert / update 相当）

4-9. 重複除去（ウォーターマーク内）

4-10. COPY INTO（SQL, 冪等な一括/定期ロード）

5. 試験で問われるポイント

6. 理解度チェックリスト