Associate 学習教材 ④ Lakeflow Jobs（オーケストレーション）

Databricks 公式ドキュメント（日本語版）の内容を、重要用語・概念を漏らさずまとめた自習用教材です。参照した公式ページ:
Lakeflow ジョブ（概要）: https://learn.microsoft.com/ja-jp/azure/databricks/jobs/
チュートリアル: Lakeflow パイプラインで ETL パイプラインを構築する: https://learn.microsoft.com/ja-jp/azure/databricks/getting-started/data-pipeline-get-started
パイプラインにデータを読み込む（旧 dlt/load → 現 ldp/load）: https://learn.microsoft.com/ja-jp/azure/databricks/ldp/load
チュートリアル: Lakeflow Pipelines エディターで最初のパイプラインを作成する: https://learn.microsoft.com/ja-jp/azure/databricks/ldp/tutorial-get-started
（補足）スケジュールとトリガーでジョブを自動化する: https://learn.microsoft.com/ja-jp/azure/databricks/jobs/triggers
（補足）スケジュールに従ってジョブを実行する: https://learn.microsoft.com/ja-jp/azure/databricks/jobs/scheduled
（補足）期待値でデータ品質を管理する: https://learn.microsoft.com/ja-jp/azure/databricks/ldp/expectations
※ 用語の呼称について: 「Lakeflow ジョブ（Lakeflow Jobs）」は旧称 Databricks Workflows / Jobs、「Lakeflow 宣言型パイプライン / Spark 宣言型パイプライン（Lakeflow Declarative Pipelines）」は旧称 Delta Live Tables（DLT） です。本教材では現行の呼称を主に使い、旧称を併記します。

1. このセクションの概要

このセクションは、Databricks 上でデータ処理を 自動化・スケジュール・調整（オーケストレーション） する仕組みを扱います。中心となるのは次の 2 つのプロダクトで、両者は組み合わせて使います。

Lakeflow ジョブ（Lakeflow Jobs、旧 Databricks Workflows / Jobs） 複数の タスク（task） を 依存関係（dependencies） に沿って実行順序どおりに調整し、トリガー（trigger） やスケジュールで自動起動する ワークフロー自動化（workflow automation） の仕組み。ETL、ノートブック実行、機械学習（ML）、dbt や外部システム連携などをオーケストレーションします。
Lakeflow 宣言型パイプライン（Lakeflow / Spark Declarative Pipelines、旧 Delta Live Tables / DLT） 「どんなテーブルを作りたいか」を 宣言的（declarative） に定義すると、Databricks が依存関係の解決・実行・データ品質チェックを自動で行う ETL フレームワーク。ストリーミングテーブル（streaming table）、マテリアライズドビュー（materialized view）、Auto Loader（自動ローダー）、Expectations（期待値／データ品質制約） が主役になります。

大きな流れは「パイプラインでデータを取り込み・変換 → そのパイプラインをジョブでスケジュール実行」です。ジョブは「いつ・どの順で動かすか」（オーケストレーション層）、パイプラインは「どんなテーブルを・どんな品質で作るか」（データ変換層）を担当する、という役割分担を理解することが本セクションの要点です。

2. 重要用語集

用語（日本語）	English	説明
Lakeflow ジョブ	Lakeflow Jobs（旧 Databricks Workflows / Jobs）	複数タスクを調整・スケジュール・実行するための主要リソース。ワークフロー自動化の単位。
ジョブ	Job	オーケストレーションの主リソース。1 つのタスクから、条件分岐・依存を持つ数百タスクまで構成可能。
タスク	Task	ジョブ内の個々の作業単位。ノートブック、パイプライン、Python スクリプトなど種類がある。
依存関係	Dependency	タスク間の実行順序の関係。あるタスクを他タスクの完了後に実行したり、条件付きで実行したりできる。
有向非巡回グラフ	DAG（Directed Acyclic Graph）	ジョブ内のタスク（またはパイプライン内のテーブル）の依存関係を視覚化したグラフ。
トリガー	Trigger	特定の条件・イベントでジョブ実行を開始する仕組み。時間ベース／イベントベースがある。
スケジュール（cron）	Schedule / Cron	時間ベースのトリガー。単純スケジュール（間隔指定）と詳細スケジュール（Quartz cron 構文）がある。
Quartz cron 構文	Quartz cron syntax	詳細スケジュールで使う cron 式の記法。UI の「Cron 構文の表示」で編集可能。
連続（継続的）トリガー	Continuous	ジョブを常時実行し続ける。実行が完了・失敗するたびに次の実行を起動する。
ファイル到着トリガー	File arrival trigger	監視対象の Unity Catalog の場所に新しいファイルが届いたときにジョブを起動。
テーブル更新トリガー	Table update trigger	ソーステーブルが更新されたときにジョブを起動。
モデル更新トリガー	Model update trigger	Unity Catalog モデルの作成・バージョン準備完了・エイリアス設定（ベータ）時に起動。
手動トリガー	Manual（None）	「今すぐ実行」ボタンや外部ツールから手動／プログラム的に起動。
リトライ（再実行）	Retry / Repair	失敗・スキップしたタスクのサブセットのみを再実行できる（Repair run）。
通知	Notification	ジョブの失敗時・実行が長すぎる時などに送るメール／Webhook（Slack など）。
パラメータ	Parameter	実行時にジョブ内のタスクへ自動的にプッシュされる実行時値。
同時実行の最大数	Max concurrent runs	既定では 1 度に 1 実行。詳細設定で上限を引き上げ可能。超過分の実行はスキップ。
Lakeflow 宣言型パイプライン	Lakeflow / Spark Declarative Pipelines（旧 Delta Live Tables / DLT）	宣言的にデータセットを定義する ETL フレームワーク。依存解決・実行・品質管理を自動化。
パイプライン	Pipeline	宣言型パイプラインの実行単位。ソースコードで定義したデータセット群を更新する。
ソースコード	Source code	パイプラインのデータセット（テーブル・ビュー）を定義するファイル。1 ファイル 1 言語。
ストリーミングテーブル	Streaming table	追加専用ソースを増分的・冪等に取り込むテーブル。取り込みの主要データセット型。
マテリアライズドビュー	Materialized view	クエリ結果を実体化して保持するビュー。変換・集計結果の保存に使う。
Auto Loader（自動ローダー）	Auto Loader（`cloudFiles`）	クラウドオブジェクトストレージに到着した新規ファイルを自動検出し増分取り込みする機能。
期待値（データ品質制約）	Expectations	クエリを通る各レコードに SQL ブール制約を適用するデータ品質チェック。
制約	Constraint	期待値の中身。各レコードが true/false で評価される SQL 条件。
ON VIOLATION	ON VIOLATION	制約違反時の動作を指定する句（DROP ROW / FAIL UPDATE）。
read_files	`read_files()`	SQL でファイルを直接読み込む関数。STREAM と組み合わせてストリーミング取り込み。
Unity Catalog	Unity Catalog	ガバナンス層。パイプライン利用や外部の場所での認証に必要。
ボリューム	Volume	Unity Catalog 管理下のファイル格納領域。CSV 等の取り込み元にできる。
外部の場所	External location	Auto Loader がクラウドストレージ認証に使う Unity Catalog オブジェクト。`READ FILES` 権限が必要。
サーバーレスコンピューティング	Serverless compute	パイプライン／ジョブを動かすマネージド計算。チュートリアルで推奨。地域制限あり。
宣言型オートメーションバンドル	Databricks Asset Bundles（DAB）	パイプライン／ジョブをコードで管理し CI/CD・バージョン管理する仕組み。

3. 詳細解説

3-1. Lakeflow ジョブとは（タスク・依存関係・オーケストレーション）

Lakeflow ジョブ（Lakeflow Jobs、旧 Databricks Workflows / Jobs） は、Databricks 上でデータ処理ワークロードをオーケストレーションする ワークフロー自動化 の仕組みです。複数のタスクを大規模ワークフローの一部として調整・実行し、繰り返し行う頻繁なタスクを最適化・スケジュールし、複雑なワークフローを管理できます。

ジョブとは Databricks ではワークフロー内のタスクをスケジュール・調整するためにジョブを使います。一般的な用途は次のとおりです。

ETL ワークフロー（データの読み込み・変換）
ノートブックの実行
機械学習（ML）ワークフロー（モデルの構築・トレーニング・デプロイを制御された反復可能な方法で行う）
dbt や Azure Data Factory（ADF）など外部システムとの統合

ジョブは 1 つ以上のタスク で構成され、ビジュアル作成 UI を使った 分岐（if/else ステートメント） や ループ（For each） などのカスタム制御フローロジックをサポートします。

オーケストレーションの 3 つの主要概念

ジョブ（Job） — 操作の調整・スケジュール・実行のための主要リソース。1 タスクから、条件付きロジックと依存関係を持つ何百ものタスクまで複雑さはさまざま。ジョブ内のタスクは 有向非巡回グラフ（DAG） で視覚的に表されます。ジョブのプロパティとして次を指定できます。
- トリガー（Trigger）: ジョブを実行するタイミングを定義。
- パラメータ（Parameter）: ジョブ内のタスクに自動的にプッシュされる実行時パラメータ。
- 通知（Notification）: ジョブ失敗時や実行が長すぎるときに送るメール／Webhook。
- Git: ジョブタスクのソース管理設定。
タスク（Task） — ジョブ内の特定の作業単位。タスクは他タスクに依存したり、条件付きで他タスクを実行したりできるため、条件付きロジックと依存関係を持つ複雑なワークフローを作成できます。主なタスク種類の例:
- ノートブックタスク（Notebook task）: Databricks ノートブックを実行。ノートブックへのパスと必要パラメータを指定。
- パイプラインタスク（Pipeline task）: パイプラインを実行。マテリアライズドビューやストリーミングテーブルを含む既存の Lakeflow パイプラインを指定。
- Python スクリプトタスク（Python script task）: Python ファイルを実行。ファイルパスと必要パラメータを指定。
- （このほか多数の種類があり、完全な一覧は「タスクの種類」ドキュメントを参照。）
トリガー（Trigger） — 特定の条件・イベントに基づいてジョブ実行を開始する仕組み。
- 時間ベース: スケジュールされた時刻に実行（例: 毎日午前 2 時）。
- イベントベース: クラウドストレージに新しいデータが到着したときに実行、など。

例: 毎日のデータ処理・検証ジョブ（公式ドキュメントの例）

最初のタスクで収益データを取り込む。
2 番目のタスクは null をチェックする if/else。
null でなければ変換タスクを実行。
null であればデータ品質検証を使うノートブックタスクを実行。
毎日同じ時刻に実行するようスケジュール。

監視と可観測性（monitoring / observability） ジョブには監視・可観測性のサポートが組み込まれています。

UI でのジョブ監視: 所有者、最後の実行結果などの詳細を表示。プロパティでフィルタ、実行履歴の表示、各タスクの詳細情報の取得が可能。
実行状態とメトリック: ジョブ実行の成否、各タスクのログとメトリックを報告し、問題診断とパフォーマンス把握に使う。
通知とアラート: メール、Slack、カスタム Webhook などでジョブイベントの通知を設定。
システムテーブル経由のクエリ: アカウント全体のジョブ実行・タスクを記録するシステムテーブルを使い、パフォーマンスやコストを分析、ダッシュボードで可視化。

制限事項（覚えておくべき上限）

1 ワークスペースでのタスク同時実行は 2,000 に制限。すぐ開始できない実行を要求すると 429 Too Many Requests が返る。
1 時間に 1 ワークスペースで作成できるジョブ数は 10,000（「実行の送信（run submit）」を含む）。REST API・ノートブックワークフロー経由も対象。
1 ワークスペースに保存できるジョブは最大 10,000（※「保存されたパイプライン」の上限 12,000 と混同しないこと）。
1 ジョブに含められるタスクは最大 1,000。
タスクがパラメータに動的値を使う場合、ジョブパラメータは 10,000 文字 に制限。

プログラムによる管理（外部連携） ワークフローはプログラムからスケジュール・調整できます。

Databricks CLI、宣言型オートメーションバンドル（Databricks Asset Bundles）、VS Code 用 Databricks 拡張機能、Databricks SDK、ジョブ REST API。
外部オーケストレーションツールとして Apache Airflow（ワークフローを Python ファイルで定義）や Azure Data Factory（ADF） も利用可能。ADF は ADF パイプライン内で Databricks ノートブック・Python スクリプト・JAR パッケージ化コードを実行できる。

3-2. ジョブのトリガー・スケジュール・リトライ・通知・パラメータ

トリガーの種類（Trigger types） Lakeflow ジョブは次の状況で自動起動するよう構成できます。

トリガーの種類	English	動作
スケジュール済み	Scheduled	時間ベースのスケジュールに基づいてジョブ実行をトリガー。
テーブルの更新	Table update	ソーステーブルが更新されたときにトリガー。
ファイルの到着	File arrival	監視対象の Unity Catalog の保存場所に新しいファイルが到着したときにトリガー。
モデルの更新	Model update	Unity Catalog モデルの作成／モデルバージョン準備完了／モデルエイリアス設定（ベータ）時にトリガー。
継続的（連続）	Continuous	ジョブを常に実行し続ける。実行が完了・失敗するたびに次の実行を起動。
なし（手動）	None（Manual）	「今すぐ実行」ボタンや他のオーケストレーションツールから手動／プログラム的にトリガー。

トリガーの構成手順: ジョブを開く → [ジョブの詳細] の [スケジュールとトリガー] → [トリガーの追加] → 種類（スケジュール／テーブルの更新／ファイルの到着／モデルの更新／継続的）を選択 → [保存]。保存すると構成に従い自動実行される。

既存トリガーの管理: [スケジュールとトリガー] セクションから 編集・一時停止（Pause）・再開（Resume）・削除 が可能。一時停止すると新しい実行は開始されないが、実行中のものは継続する。[アクティブ] と [一時停止] を切り替えてトリガーの状態も制御できる。継続的トリガーを再開したとき実行がアクティブなら、その実行完了まで待って次を起動する。

スケジュールトリガー（cron）の詳細 スケジュールされたトリガーには 2 つのオプションがあります。

単純スケジュール（Simple）: 時間単位と間隔を指定して定期実行（例: 最初の実行から 12 時間ごと）。最初の実行時刻は指定できず、スケジューラが選ぶ。
詳細スケジュール（Advanced）: 期間・開始時刻・タイムゾーンを細かく制御。[Cron 構文の表示] にチェックを入れると Quartz cron 構文 でスケジュールを表示・編集できる。

スケジュールに関する注意点:

cron 式の構成に関わらず、Databricks はスケジュールでトリガーされる後続実行の間に 最小 10 秒 の間隔を強制する。
タイムゾーン: 夏時間（DST）を監視するタイムゾーンを選ぶと、1 時間ごとのジョブがスキップされたり 1〜2 時間遅れて見えることがある。絶対時間で毎時実行したい場合は UTC を選ぶ。
ジョブスケジューラは低遅延ジョブを想定していない。ネットワークやクラウドの問題で実行が最大数分遅れることがあり、その場合サービス復帰後すぐに実行される。

同時実行（concurrency） 既定では 1 度にアクティブにできるジョブ実行は 1 つだけ。詳細設定で最大同時実行数を引き上げられる。構成した最大コンカレンシーを超えた実行は スキップ される。

リトライ（再実行 / Repair run） 複数タスクを含むジョブで一部のタスクが成功しなかった場合、失敗したタスクのサブセットのみを再実行 できる（Repair run）。全体をやり直す必要はない。

通知（Notification） ジョブが失敗したとき、または実行が長すぎるときに、メール・Slack・カスタム Webhook などで通知を送れる。UI のスケジュール作成時に「その他のオプション」から通知を設定できる。

パラメータ（Parameter） ジョブに設定した実行時パラメータは、ジョブ内のタスクへ自動的にプッシュされる。タスクが動的値を使う場合、ジョブパラメータは 10,000 文字までという制限に注意。

3-3. Lakeflow 宣言型パイプライン（旧 Delta Live Tables）の概要

Lakeflow 宣言型パイプライン（Lakeflow / Spark Declarative Pipelines、旧 Delta Live Tables / DLT） は、ETL を宣言的に構築するためのフレームワークです。ソースからデータを読み取り、要件（データ品質チェックやレコードの重複除去など）に基づいて変換し、データウェアハウスやデータレイクなどのターゲットに書き込む手順を実装します。

パイプラインの構成要素

ソースコード（source code）: パイプライン構文でデータセット（テーブル・ビュー）を定義するファイル。各ファイルは 1 言語のみ（Python または SQL）だが、言語ごとのファイルを複数まとめて 1 つのパイプラインにできる。既定では transformations フォルダー内のすべての *.py / *.sql ファイルがパイプラインのソースの一部になる。
パイプライングラフ（pipeline graph）: コードから自動生成される依存関係の DAG。どのテーブルがどのテーブルのソースかを可視化する。

2 つの主要なデータセット型

ストリーミングテーブル（streaming table）: 追加専用ソースを 増分的（incremental）かつ冪等（idempotent） に取り込むテーブル。取り込み（インジェスト）の主要データセット型。Auto Loader でクラウドオブジェクトストレージから、または Kafka などのメッセージバスから取り込む場合に便利。既定では 追加専用ソース を要求する。
マテリアライズドビュー（materialized view）: クエリ結果を実体化して保持するビュー。変換・検証・集計した結果の保存に使う。

Lakeflow Pipelines エディター パイプラインのソースコードを対話的に開発・検証するエディター。主な機能:

選択的な実行（selective run）: ファイル単位／パイプライン単位で更新を実行。
データプレビュー（data previews）: 生成テーブルのデータを確認。
対話型パイプライングラフ（interactive pipeline graph, DAG）: データセットの依存を視覚化。
パイプライン資産ブラウザー（pipeline assets browser）: transformations（本体）、explorations（アドホック分析用。既定ではパイプライン更新の一部として実行されない）などのフォルダー構成を管理。
宣言型オートメーションバンドル（Databricks Asset Bundles）統合: バージョン管理・CI/CD・ソース管理パイプライン化を直接行える。

パイプライン作成の流れ（チュートリアルより）

サイドバーの [新規（New）] → [ETL パイプライン] を選択すると、New Pipeline <date> <time> のような既定名でパイプラインエディターが開く。
（任意）名前を付ける／カタログ・スキーマの既定値を設定する。
my_transformation ソースファイルの言語（Python / SQL）を選ぶ。
[サンプルコードを使用] でサンプルが transformations フォルダーに表示される。
[ファイルの実行] または [パイプラインの実行] で更新を開始。更新が完了するとエディターがパイプライン情報（DAG、更新概要、テーブル詳細）で更新される。

前提条件（チュートリアル）: Unity Catalog を有効化、サーバーレスコンピューティングを有効化（サーバーレスのパイプラインは全リージョンで使えるわけではない）、コンピューティングリソースの作成・アクセス権限、スキーマ作成権限（USE CATALOG と CREATE SCHEMA、または ALL PRIVILEGES）、ボリューム作成権限（USE SCHEMA と CREATE VOLUME）など。

3-4. パイプラインでのデータ読み込みと Expectations（データ品質チェック）

データの読み込み（ingestion / load）

パイプラインでは、Databricks 上の Apache Spark がサポートする任意のデータソースからデータを読み込めます。Spark DataFrame（ストリーミング DataFrame や pandas on Spark を含む）を返すクエリでデータセットを定義します。データインジェストのほとんどのユースケースでは、ストリーミングテーブルの使用が推奨 されます。

主な読み込みパターン:

既存テーブルからの読み込み: spark.read.table("...")（Python）／FROM <table>（SQL）。変換や集計に使う。
クラウドオブジェクトストレージ／ボリュームからのファイル読み込み: Auto Loader（自動ローダー、cloudFiles フォーマット） を推奨。到着し続けるファイルを 増分的・冪等 に読み込む。SQL では read_files() 関数を STREAM と組み合わせて使う。
- 注意: ファイル通知モード（file notification） の Auto Loader でパイプライン／ストリーミングテーブルの 完全更新（full refresh） を行う場合、リソースの手動クリーンアップが必要（CloudFilesResourceManager を使用）。
- 注意: Unity Catalog 有効パイプライン で Auto Loader を使うには 外部の場所（external location） が必要。読み取るストレージパスに外部の場所を構成し、実行ユーザーに READ FILES 権限を付与する。
メッセージバスからの読み込み: Kafka（read_kafka）、Google Pub/Sub（read_pubsub）、Kinesis（read_kinesis）、Pulsar（read_pulsar）など。低遅延読み込みには継続実行と拡張自動スケーリングのストリーミングテーブルを推奨。
Azure Event Hubs: Apache Kafka 互換インターフェイスを提供。パイプラインランタイム同梱の Structured Streaming Kafka コネクタで読み込む。
外部システム: Lakehouse Federation（Databricks Runtime 13.3 LTS 以上、プレビューチャネル構成が必要）、または Python でソース（例: PostgreSQL）から取り込む。SQL 非対応のソースは Python を使い、同一パイプラインに Python と SQL のファイルを混在できる。
小規模／静的データセット: Apache Spark の読み込み構文（spark.read.format("json") など）でバッチ読み込み。サポートされる全ファイル形式が使える。
Python カスタムデータソース: spark.dataSource.register で登録し、バッチ・ストリーミング両方で読み取れる。

その他の重要ポイント:

skipChangeCommits: ストリーミングテーブルは既定で追加専用ソースを要求する。ソースで更新・削除がある場合（例: GDPR の「忘れられる権利」）は spark.readStream.option("skipChangeCommits", "true") でそれらの変更を無視できる。この設定は option() でのみ機能し、create_auto_cdc_flow() のターゲットには使えない。
シークレット（secrets）: アクセスキーやパスワードは Databricks シークレットに格納し、パイプライン設定のクラスター構成の Spark プロパティで参照する。spark.hadoop. のキーには spark_conf プレフィックスが必要（{{secrets/<scope>/<secret>}} 形式で参照）。
read_files SQL 関数 は全 SQL 環境で共通。SQL でファイルへ直接アクセスする場合の推奨パターン。

Expectations（期待値／データ品質制約）

Expectations（期待値） は、パイプラインのマテリアライズドビュー・ストリーミングテーブル・ビュー作成ステートメントに付ける任意の句で、クエリを通過する 各レコード にデータ品質チェックを適用します。標準の SQL ブールステートメントで制約を指定し、1 つのデータセットに複数の期待値を組み合わせられます。Databricks SQL で作成したスタンドアロンパイプラインの CREATE STREAMING TABLE / CREATE MATERIALIZED VIEW でも CONSTRAINT expectation_name EXPECT (expr) 句として定義できます。

期待値の 3 つの構成要素

期待値の名前（expectation name）: 追跡・監視の識別子。特定のデータセット内で一意 である必要がある（別データセット間では再利用可）。検証内容を伝える名前を付ける。
評価する制約（constraint）: 各レコードで true/false に評価される SQL 条件。有効な SQL 構文が必要で、カスタム Python 関数・外部サービス呼び出し・他テーブルを参照するサブクエリは使えない。
無効なレコードに対するアクション（action on violation）: 制約違反時の動作を指定する。

無効レコードへの 3 つのアクション

アクション	SQL 構文	Python 構文	結果
warn（既定）	`EXPECT`	`dp.expect`	無効レコードもターゲットに書き込む（保持）。違反数などのメトリックを収集。
drop（削除）	`EXPECT ... ON VIOLATION DROP ROW`	`dp.expect_or_drop`	無効レコードは書き込み前に削除。削除数はメトリックとしてログに記録。
fail（失敗）	`EXPECT ... ON VIOLATION FAIL UPDATE`	`dp.expect_or_fail`	無効レコードがあると更新が失敗。再処理には手動介入が必要。テーブル更新の場合はトランザクションをアトミックにロールバック。

fail の挙動の注意:

トリガーされたパイプライン（triggered pipeline） では 1 フローの障害で他の並列フローは失敗しない。
継続的パイプライン（continuous pipeline） では、予期されたエラーでフローが停止し、依存する全フローが停止、停止理由のメッセージが出力される。
期待違反で失敗したら、パイプラインを再実行する前にコードを修正して無効データを正しく処理する必要がある。専用のエラーメッセージ（例: [EXPECTATION_VIOLATION.VERBOSITY_ALL] ...）で違反レコードを特定できる。

複数の期待値をまとめて適用（Python のみ） Python では複数の期待値を 辞書（dict） にまとめ、集合アクションを指定できる（SQL は 1 データセットに複数期待値を書けるがグループ化・集合アクションは不可）。

expect_all … 保持（warn）
expect_all_or_drop … 削除
expect_all_or_fail … 失敗

期待値のメトリック・追跡

パイプライン UI の [データ品質（Data quality）] タブで warn / drop のメトリックを確認できる（fail は更新が失敗するためメトリックは記録されない）。
期待値をサポートするのは ストリーミングテーブル・マテリアライズドビュー・一時ビュー のみ。シンク（sink）など一部フロー種類、AUTO CDC FROM SNAPSHOT は期待値をサポートしない。
Databricks SQL のスタンドアロンパイプラインでは UI の [データ品質] タブは使えず、イベントログ（event log） をクエリしてメトリックを確認する。

3-5. ジョブとパイプラインの関係・使い分け

役割分担のイメージ

観点	Lakeflow 宣言型パイプライン（旧 DLT）	Lakeflow ジョブ（旧 Workflows）
主な役割	データの取り込み・変換・品質管理（ETL の「中身」）	複数処理の調整・スケジュール・自動化（オーケストレーション）
定義の仕方	宣言的（作りたいテーブルを定義、依存は自動解決）	タスクと依存関係を組んで実行順を制御（DAG）
主な構成要素	ストリーミングテーブル、マテリアライズドビュー、Expectations、Auto Loader	タスク（ノートブック／パイプライン／Python など）、トリガー、通知、パラメータ
データ品質	Expectations で組み込み対応	タスクとして品質検証ノートブックを組み込む等

両者の連携 パイプラインは ジョブのタスク（パイプラインタスク） として組み込める。実際のワークフローでは「パイプラインでデータを取り込み・変換 → そのパイプラインをジョブでスケジュール実行」という形が基本。

チュートリアルでも、ETL パイプラインを作った後に [スケジュール] → [スケジュールの追加] → 新しいスケジュール で「パイプラインを実行するジョブ」を作成し、既定では 1 日 1 回実行にして、データ取り込み・処理・分析を自動化している。作成したジョブは [ジョブとパイプライン] 一覧に現れ、実行履歴の確認や [今すぐ実行] での即時実行ができる。

使い分けの指針

「テーブルをどう作るか・データ品質をどう保つか」→ 宣言型パイプライン。
「いつ・どんな順で・何をまとめて動かすか（複数パイプラインやノートブック・ML の連携、条件分岐、通知、リトライ）」→ ジョブ。

4. 構文・コード例

4-1. ストリーミングテーブルで生データを取り込む（Auto Loader）

Python

python

from pyspark import pipelines as dp
from pyspark.sql.functions import *
from pyspark.sql.types import DoubleType, IntegerType, StringType, StructType, StructField

file_path = f"/databricks-datasets/songs/data-001/"

schema = StructType([
    StructField("artist_id", StringType(), True),
    StructField("artist_name", StringType(), True),
    StructField("duration", DoubleType(), True),
    StructField("title", StringType(), True),
    StructField("year", IntegerType(), True),
    # ... 実際は全カラムを定義 ...
])

# ボリューム/クラウドストレージから増分取り込みするストリーミングテーブル
@dp.table(
  comment="Raw data from a subset of the Million Song Dataset."
)
def songs_raw():
  return (spark.readStream
    .format("cloudFiles")            # Auto Loader
    .schema(schema)
    .option("cloudFiles.format", "csv")
    .option("sep", "\t")
    .load(file_path))

SQL

sql

CREATE OR REFRESH STREAMING TABLE songs_raw
COMMENT "Raw data from a subset of the Million Song Dataset."
AS SELECT *
FROM STREAM read_files(
  '/databricks-datasets/songs/data-001/part*',
  format => "csv",
  header => "false",
  delimiter => "\t",
  schema => """
    artist_id STRING,
    artist_name STRING,
    duration DOUBLE,
    title STRING,
    year INT
  """,
  schemaEvolutionMode => "none");

4-2. マテリアライズドビューで変換 + Expectations（品質チェック）

Python

python

@dp.materialized_view(
  comment="Million Song Dataset with data cleaned and prepared for analysis."
)
@dp.expect("valid_artist_name", "artist_name IS NOT NULL")   # warn: 保持
@dp.expect("valid_title", "song_title IS NOT NULL")
@dp.expect("valid_duration", "duration > 0")
def songs_prepared():
  return (
    spark.read.table("songs_raw")
      .withColumnRenamed("title", "song_title")
      .select("artist_id", "artist_name", "duration", "release",
              "tempo", "time_signature", "song_title", "year")
  )

SQL

sql

CREATE OR REFRESH MATERIALIZED VIEW songs_prepared(
  CONSTRAINT valid_artist_name EXPECT (artist_name IS NOT NULL),
  CONSTRAINT valid_title       EXPECT (song_title IS NOT NULL),
  CONSTRAINT valid_duration    EXPECT (duration > 0)
)
COMMENT "Million Song Dataset with data cleaned and prepared for analysis."
AS SELECT artist_id, artist_name, duration, release, tempo, time_signature,
          title AS song_title, year
FROM songs_raw;

4-3. 集計・並べ替えのマテリアライズドビュー

SQL

sql

CREATE OR REFRESH MATERIALIZED VIEW top_artists_by_year
COMMENT "A table summarizing counts of songs released by the artists each year."
AS SELECT
  artist_name,
  year,
  COUNT(*) AS total_number_of_songs
FROM songs_prepared
WHERE year > 0
GROUP BY artist_name, year
ORDER BY total_number_of_songs DESC, year DESC;

4-4. ON VIOLATION の 3 パターン

sql

-- warn（既定）: 違反レコードも保持
CONSTRAINT valid_timestamp EXPECT (timestamp > '2012-01-01')

-- drop: 違反レコードを削除
CONSTRAINT valid_current_page
  EXPECT (current_page_id IS NOT NULL AND current_page_title IS NOT NULL)
  ON VIOLATION DROP ROW

-- fail: 違反があれば更新を失敗（ロールバック）
CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE

python

# warn / drop / fail（Python）
@dp.expect("valid timestamp", "timestamp > '2012-01-01'")
@dp.expect_or_drop("valid_current_page",
                   "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")
@dp.expect_or_fail("valid_count", "count > 0")

4-5. 複数の期待値をまとめて適用（Python のみ）

python

valid_pages = {
  "valid_count": "count > 0",
  "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"
}

@dp.table
@dp.expect_all(valid_pages)          # 保持
def raw_data():
  ...

@dp.table
@dp.expect_all_or_drop(valid_pages)  # 削除
def prepared_data():
  ...

@dp.table
@dp.expect_all_or_fail(valid_pages)  # 失敗
def customer_facing_data():
  ...

4-6. データ品質チェックで無効行を削除する（チュートリアル）

SQL

sql

-- email が無いレコードを削除
CREATE MATERIALIZED VIEW users_cleaned
(
  CONSTRAINT non_null_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW
) AS
SELECT *
FROM sample_users_<date_time>;

Python

python

from pyspark import pipelines as dp

@dp.materialized_view
@dp.expect_or_drop("no null emails", "email IS NOT NULL")
def users_cleaned():
    return spark.read.table("sample_users_<date_time>")

4-7. Kafka / Pub/Sub からの取り込み

python

# Kafka
@dp.table
def kafka_raw():
  return (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka_server:9092")
    .option("subscribe", "topic1")
    .load())

sql

-- Kafka（SQL）
CREATE OR REFRESH STREAMING TABLE kafka_raw AS
  SELECT * FROM STREAM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'topic1'
  );

4-8. アドホック分析クエリ（explorations、パイプライン更新には含まれない）

sql

-- 1990 年以降、各年で最も曲をリリースしたアーティスト
SELECT artist_name, total_number_of_songs, year
  FROM <catalog>.<schema>.top_artists_by_year
  WHERE year >= 1990
  ORDER BY total_number_of_songs DESC, year DESC;

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

ジョブ / タスク / トリガーの 3 概念 を区別できること。ジョブ=主リソース、タスク=作業単位、トリガー=起動条件。タスクの依存関係は DAG で表される。
タスクの種類: ノートブック、パイプライン、Python スクリプトなど。パイプラインをタスクとしてジョブに組み込める（パイプラインタスク）。
トリガーの種類: スケジュール済み／テーブル更新／ファイル到着／モデル更新／継続的／手動（None）。イベントベース（ファイル到着など）と時間ベースの区別。
スケジュール: 単純スケジュール（間隔指定）と詳細スケジュール（Quartz cron 構文）。後続実行間に 最小 10 秒 の間隔。絶対時間の毎時実行は UTC を選ぶ（DST の落とし穴）。
同時実行: 既定は 1 実行のみ。上限超過分はスキップ。
リトライ: 失敗タスクのサブセットのみ再実行できる（Repair run）。全体をやり直さない。
通知: メール・Slack・Webhook を失敗時や長時間実行時に送れる。
宣言型パイプライン の 2 大データセット型 = ストリーミングテーブル（増分・冪等な取り込み、追加専用ソース）と マテリアライズドビュー（実体化した変換結果）。
Auto Loader（cloudFiles） = クラウドストレージの新規ファイルを自動検出・増分取り込み。SQL では read_files() + STREAM。
Expectations の 3 アクションを正確に: EXPECT（warn=保持・既定）／ON VIOLATION DROP ROW（削除）／ON VIOLATION FAIL UPDATE（失敗・ロールバック）。Python では expect / expect_or_drop / expect_or_fail、複数まとめは expect_all(_or_drop/_or_fail)（Python のみ）。
期待値の制約は 有効な SQL のみ（Python UDF・外部呼び出し・他テーブル参照サブクエリは不可）。名前はデータセット内で一意。
ジョブとパイプラインの使い分け: 変換・品質=パイプライン、順序制御・スケジュール・自動化=ジョブ。パイプラインは [スケジュール] からジョブ化してスケジュール実行できる。
前提: Unity Catalog 有効化、Auto Loader での外部の場所と READ FILES 権限、サーバーレスは地域制限あり。
主な上限値: タスク同時実行 2,000、1 時間あたりジョブ作成 10,000、保存ジョブ 10,000、1 ジョブ最大 1,000 タスク。

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

[ ] Lakeflow ジョブ（旧 Workflows）の「ジョブ・タスク・トリガー」の 3 概念をそれぞれ説明できる
[ ] タスクの依存関係が DAG で表現されることと、条件分岐（if/else）・ループが使えることを理解している
[ ] タスクの主な種類（ノートブック／パイプライン／Python スクリプト）を挙げられる
[ ] トリガーの 6 種類（スケジュール／テーブル更新／ファイル到着／モデル更新／継続的／手動）を区別できる
[ ] 単純スケジュールと詳細スケジュール（Quartz cron 構文）の違い、最小 10 秒間隔、UTC と DST の注意点を説明できる
[ ] 既定の同時実行数（1）と、上限超過分がスキップされることを理解している
[ ] Repair run（失敗タスクのサブセット再実行）と通知（メール／Slack／Webhook）の役割を説明できる
[ ] ジョブパラメータが実行時にタスクへプッシュされることを理解している
[ ] Lakeflow 宣言型パイプライン（旧 DLT）が宣言的 ETL であり、依存を自動解決することを説明できる
[ ] ストリーミングテーブルとマテリアライズドビューの違いと用途を説明できる
[ ] Auto Loader（cloudFiles）が新規ファイルを増分・冪等に取り込むこと、SQL では read_files() + STREAM を使うことを理解している
[ ] Unity Catalog パイプラインで Auto Loader を使うには外部の場所と READ FILES 権限が必要だと知っている
[ ] Expectations の 3 アクション（warn=保持／drop=削除／fail=失敗）を SQL・Python 両方の構文で書ける
[ ] ON VIOLATION DROP ROW と ON VIOLATION FAIL UPDATE、expect_or_drop / expect_or_fail を対応づけられる
[ ] 複数期待値のグループ適用（expect_all 系）が Python のみであることを理解している
[ ] 期待値の制約に使える SQL の制限（UDF・外部呼び出し・他テーブル参照サブクエリ不可）を理解している
[ ] パイプラインをジョブのタスクとして組み込み、[スケジュール] からジョブ化してスケジュール実行する流れを説明できる
[ ] ジョブ（オーケストレーション）とパイプライン（変換・品質）の使い分けを説明できる

Associate 学習教材 ④ Lakeflow Jobs（オーケストレーション） ​

1. このセクションの概要 ​

2. 重要用語集 ​

3. 詳細解説 ​

3-1. Lakeflow ジョブとは（タスク・依存関係・オーケストレーション） ​

3-2. ジョブのトリガー・スケジュール・リトライ・通知・パラメータ ​

3-3. Lakeflow 宣言型パイプライン（旧 Delta Live Tables）の概要 ​

3-4. パイプラインでのデータ読み込みと Expectations（データ品質チェック） ​

データの読み込み（ingestion / load） ​

Expectations（期待値／データ品質制約） ​

3-5. ジョブとパイプラインの関係・使い分け ​

4. 構文・コード例 ​

4-1. ストリーミングテーブルで生データを取り込む（Auto Loader） ​

4-2. マテリアライズドビューで変換 + Expectations（品質チェック） ​

4-3. 集計・並べ替えのマテリアライズドビュー ​

4-4. ON VIOLATION の 3 パターン ​

4-5. 複数の期待値をまとめて適用（Python のみ） ​

4-6. データ品質チェックで無効行を削除する（チュートリアル） ​

4-7. Kafka / Pub/Sub からの取り込み ​

4-8. アドホック分析クエリ（explorations、パイプライン更新には含まれない） ​

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

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

Associate 学習教材 ④ Lakeflow Jobs（オーケストレーション）

1. このセクションの概要

2. 重要用語集

3. 詳細解説

3-1. Lakeflow ジョブとは（タスク・依存関係・オーケストレーション）

3-2. ジョブのトリガー・スケジュール・リトライ・通知・パラメータ

3-3. Lakeflow 宣言型パイプライン（旧 Delta Live Tables）の概要

3-4. パイプラインでのデータ読み込みと Expectations（データ品質チェック）

データの読み込み（ingestion / load）

Expectations（期待値／データ品質制約）

3-5. ジョブとパイプラインの関係・使い分け

4. 構文・コード例

4-1. ストリーミングテーブルで生データを取り込む（Auto Loader）

4-2. マテリアライズドビューで変換 + Expectations（品質チェック）

4-3. 集計・並べ替えのマテリアライズドビュー

4-4. ON VIOLATION の 3 パターン

4-5. 複数の期待値をまとめて適用（Python のみ）

4-6. データ品質チェックで無効行を削除する（チュートリアル）

4-7. Kafka / Pub/Sub からの取り込み

4-8. アドホック分析クエリ（explorations、パイプライン更新には含まれない）

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

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