databricks-pipelines

Lakeflow Spark Declarative Pipelines の開発

最初に：CLI の基本、認証、プロファイルの選択、データ探索コマンドについては、親の databricks-core スキルを使用してください。

判断ツリー

このツリーを使って、使用するデータセットの種類と機能を判断してください。同じデータセットに複数の機能を適用できます。たとえば、ストリーミングテーブルでは、取り込みに Auto Loader、複数ストリームの統合に Append Flows、データ品質の確保に Expectations を使用できます。まずデータセットの種類を選び、次に適用可能な機能を追加してください。

User request → What kind of output?
├── Intermediate/reusable logic (not persisted) → Temporary View
│   ├── Preprocessing/filtering before Auto CDC → Temporary View feeding CDC flow
│   ├── Shared intermediate streaming logic reused by multiple downstream tables
│   ├── Pipeline-private helper logic (not published to catalog)
│   └── Published to UC for external queries → Persistent View (SQL only)
├── Persisted dataset
│   ├── Source is a periodic complete snapshot that must be diffed by key → Streaming Table + Auto CDC FROM SNAPSHOT (SQL requires DBR 18.3+)
│   ├── Source is streaming/incremental/continuously growing → Streaming Table
│   │   ├── File ingestion (cloud storage, Volumes) → Auto Loader
│   │   ├── Message bus (Kafka, Kinesis, Pub/Sub, Pulsar, Event Hubs) → streaming source read
│   │   ├── Existing streaming/Delta table → streaming read from table
│   │   ├── CDC / upserts / track changes / keep latest per key / SCD Type 1 or 2 → Auto CDC
│   │   ├── Multiple sources into one table → Append Flows (NOT union)
│   │   ├── Historical backfill + live stream → one-time Append Flow + regular flow
│   │   └── Windowed aggregation with watermark → stateful streaming
│   └── Source is batch/historical/full scan → Materialized View
│       ├── Aggregation/join across full dataset (GROUP BY, SUM, COUNT, etc.)
│       ├── Gold layer aggregation from streaming table → MV with batch read (spark.read / no STREAM)
│       ├── JDBC/Federation/external batch sources
│       └── Small static file load (reference data, no streaming read)
├── Output to external system (Python only) → Sink
│   ├── Existing external table not managed by this pipeline → Sink with format="delta"
│   │   (prefer fully-qualified dataset names if the pipeline should own the table — see Publishing Modes)
│   ├── Kafka / Event Hubs → Sink with format="kafka" + @dp.append_flow(target="sink_name")
│   ├── Custom destination not natively supported → Sink with custom format
│   ├── Custom merge/upsert logic per batch → ForEachBatch Sink (Public Preview)
│   └── Multiple destinations per batch → ForEachBatch Sink (Public Preview)
└── Data quality constraints → Expectations (on any dataset type)

よくある落とし穴

• 名称 → SDP ＝ LDP ＝ Lakeflow Declarative Pipelines ＝（旧称）DLT です。ユーザーがこれらの名称を使った場合、すべて同じものを指します。
• 種類を指定せずに「テーブルを作成」と言われた場合 → ソースがストリーミングかバッチかを確認してください。ストリーミングソースならストリーミングテーブル、バッチソースならマテリアライズドビューを使用します。組み合わせが適切でないと、検証時にエラーになります。
• ストリーミングソースに対する集計 → バッチ読み取り（spark.read.table／STREAM を指定しない SELECT FROM）を使用するマテリアライズドビューを使ってください。ストリーミングテーブルは追記専用であり、ソースの行が変更されても集計を再計算しません。マテリアライズドビューは再計算します。
• 中間ロジック → デフォルトでは一時ビューを使用してください。複数の下流テーブルで共有ロジックを再利用する場合も同様です、一度マテリアライズすることで再処理を大幅に削減できる場合に限り、プライベートなマテリアライズドビュー／ストリーミングテーブル（private=True／CREATE PRIVATE ...）を使用してください。Auto CDC の前処理には一時ビューが必須です。CDC フローは STREAM(view_name)（SQL）または spark.readStream.table("view_name")（Python）から読み取るためです。
• ストリームのユニオン → 複数の Append Flows を使用してください。ストリーミングソース間で UNION を使うのはアンチパターンです。
• データセットの種類の変更 → ストリーミングテーブルからマテリアライズドビュー、またはマテリアライズドビューからストリーミングテーブルへ、その場で種類を変更することはできません。フルリフレッシュを行っても解決しません。既存のテーブルを手動で削除するか、新しいデータセットの名前を変更してください。
• CREATE OR REFRESH と CREATE の違い → SQL データセットではどちらも解析されますが、慣用的な規約は CREATE OR REFRESH です。プライベートデータセットの場合は、CREATE OR REFRESH PRIVATE STREAMING TABLE／... MATERIALIZED VIEW を使用してください。
• Kafka／Event Hubs シンクのシリアル化 → value 列は必須です。to_json(struct(*)) AS value を使って行をシリアル化してください。sink-python.md を参照してください。
• Auto CDC の複数列によるシーケンス指定 → SQL：SEQUENCE BY STRUCT(col1, col2)。Python：sequence_by=struct("col1", "col2")。Auto CDC のリファレンスを参照してください。
• Auto CDC の TRUNCATE（SCD タイプ 1 のみ）→ SQL：APPLY AS TRUNCATE WHEN condition。Python：apply_as_truncates=expr("condition")。truncate はサポートされていないと説明しないでください。
• 定期的な完全スナップショット → DBR 18.3 以降では SQL の AUTO CDC FROM SNAPSHOT を、古いランタイムでは Python の dp.create_auto_cdc_from_snapshot_flow を使用してください。バージョンの選択とライフサイクルのルールについては、Auto CDC のリファレンスを確認してください。
• Python 専用機能 → シンク、ForEachBatch シンク、カスタムデータソースは Python 専用です。ユーザーが SQL を使用している場合は、この点を説明し、Python への切り替えを提案してください。
• 明確な方法を 1 つ推奨する → 推奨する手順を 1 つ提示してください。アンチパターンや劣る代替案を列挙すると、混乱を招きます。トレードオフが実際に異なる場合に限り、代替案に言及してください。

よくある問題

エラー → 原因／対処方法の対応表です。エージェントが頻繁に遭遇する問題をまとめています。DAB バンドルと CLI による反復開発のデプロイに関する問題については、ワークフロー固有のリファレンスファイルを参照してください。

エラー／症状	原因／対処方法
CREATE OR REPLACE STREAMING TABLE／MATERIALIZED VIEW が拒否される	CREATE OR REPLACE は標準 SQL であり、SDP ではありません。CREATE OR REFRESH STREAMING TABLE／CREATE OR REFRESH MATERIALIZED VIEW を使用してください。
databricks fs ls /Volumes/... で CLI エラーが発生する	UC ボリュームのパスでも dbfs: プレフィックスが必要です。databricks fs ls dbfs:/Volumes/<catalog>/<schema>/<volume>/<path> を使用してください。
初回の書き込み時に DELTA_CLUSTERING_COLUMNS_DATATYPE_NOT_SUPPORTED が発生する	CLUSTER BY 列の型が BOOLEAN／ARRAY／MAP／STRUCT／BINARY です。SDP は事前検証を行わないため、送信前に DESCRIBE で確認してください。クラスタリングキーには数値／文字列／日付／タイムスタンプを使用する必要があります。型の完全なルールについては、references/performance.md を参照してください。
Cannot create streaming table from batch query	ストリーミングテーブルのクエリに、バッチ処理の FROM read_files(...) を記述しています。FROM STREAM read_files(...) を使用して Auto Loader を起動してください。
取り込み時に Column not found が発生する	schemaHints が実際のファイルスキーマと一致していません。サンプルファイルに対して DESCRIBE を実行し、ヒントを一致させてください。
ストリーミング読み取りがパーサーエラーで失敗する	ファイルの取り込みには FROM STREAM read_files(...) を、テーブル間のストリームには FROM stream(table)（または FROM STREAM table_name — 旧 DLT 形式のため、関数形式を推奨）を使用してください。混在させないでください。
サーバーレスのパイプラインが INITIALIZING のまま進まない	正常な動作です。コールドスタートのため、初回実行には数分かかります。停止しないでください。
マテリアライズドビューが増分更新されない	集計の自動増分更新には、サーバーレスとソースでの Delta 行追跡（delta.enableRowTracking = true）の両方が必要です。どちらかが欠けていると、全件再計算にフォールバックします。ユーザーが増分更新について尋ねた場合は、サーバーレスが必要であることを伝えてください。
SCD2 クエリが何も返さない／START_AT で「column not found」が発生する	Lakeflow では __START_AT／__END_AT（アンダースコア 2 つ）を使用します。現在の行は WHERE __END_AT IS NULL で指定します。
イベント出力に error.exceptions[0].message が見つからない	jq が .message（「Update X is FAILED」という内容のみ）を読み取っています。実際の原因を確認するには error.exceptions[0].message を読み取ってください。2-rapid-iteration-with-cli.md を参照してください。

パブリッシュモード

パイプラインでは、設定で構成されたデフォルトカタログとスキーマを使用します。上書きしない限り、すべてのデータセットはそこにパブリッシュされます。

• 完全修飾名：データセット名に catalog.schema.table を指定すると、パイプラインのデフォルトとは異なるカタログ／スキーマに書き込めます。パイプラインは Sink を使わずに、その場所にデータセットを直接作成します。
• USE CATALOG／USE SCHEMA：同じファイル内の以降のすべての定義について、現在のカタログ／スキーマを変更する SQL コマンドです。
• LIVE プレフィックス：非推奨です。デフォルトのパブリッシュモードでは無視されます。
• パイプライン内でデータセットを読み取る場合や定義する場合は、データセット名のみを使用してください。パイプラインですでに使用されている場合、またはユーザーが別のターゲットカタログ／スキーマを明示的に要求した場合を除き、完全修飾名は使用しないでください。

API リファレンス

機能のパイプラインコードを作成する前に、リンク先のリファレンスファイルを読んでください。 以下の各表では、機能に対応する正確な API と、その機能と言語の組み合わせに関する詳細ファイルを示しています。

一部の機能は他の機能を基盤としています。該当する場合は両方を読んでください。

• Auto Loader／Auto CDC／Sinks はストリーミングテーブルを対象とするため、streaming-table-python.md／streaming-table-sql.md も読んでください。
• Expectations はデータセットに付加するため、データセット定義ファイル（streaming-table／materialized-view／temporary-view）も読んでください。

データセット定義 API

機能	説明	Python	SQL	スキル（Py）	スキル（SQL）
ストリーミングテーブル	継続的な増分処理を行い、厳密に 1 回だけ処理し、追加のみを行います。	@dp.table() returning streaming DF	CREATE OR REFRESH STREAMING TABLE	streaming-table-python	streaming-table-sql
マテリアライズドビュー	クエリ結果を物理的に保存し、増分更新します。	@dp.materialized_view()	CREATE OR REFRESH MATERIALIZED VIEW	materialized-view-python	materialized-view-sql
一時ビュー	パイプライン内でのみ使用され、Unity Catalog には永続化されません。	@dp.temporary_view()	CREATE TEMPORARY VIEW	temporary-view-python	temporary-view-sql
永続ビュー（UC）	UC に公開され、アクセス時にクエリが実行されます（ストレージなし）。	N/A — SQL only	CREATE VIEW	—	view-sql
ストリーミングテーブル（明示的）	空のターゲットを作成し、別個のフロー（Append Flow、AUTO CDC）でデータを投入します。	dp.create_streaming_table()	CREATE OR REFRESH STREAMING TABLE (no AS)	streaming-table-python	streaming-table-sql

フローとシンクの API

機能	説明	Python	SQL	スキル（Py）	スキル（SQL）
Append Flow	ファンイン：複数のソースを 1 つのストリーミングテーブルに結合します。UNION の代わりに使用します。	@dp.append_flow()	CREATE FLOW ... INSERT INTO	streaming-table-python	streaming-table-sql
Backfill Flow	同じテーブルに対して、過去データを一度だけロードし、その後ライブストリームを継続的に取り込みます。	@dp.append_flow(once=True)	CREATE FLOW ... INSERT INTO ... ONCE	streaming-table-python	streaming-table-sql
シンク（Delta/Kafka/EH/カスタム）	ストリーミング出力を外部の Delta／Kafka／Event Hubs に書き込みます。	dp.create_sink()	該当なし — Python のみ	sink-python	—
ForEachBatch シンク	バッチごとのカスタム Python ロジック（マージ／アップサート、複数の宛先）。パブリックプレビューです。	@dp.foreach_batch_sink()	該当なし — Python のみ	foreach-batch-sink-python	—
RTM 更新フロー	リアルタイムモード：サブ秒レイテンシでフローをシンクにルーティングします。パブリックプレビューです。	@dp.update_flow(target=...)	該当なし — Python のみ	real-time-mode	—

CDC API

機能	説明	Python	SQL	スキル（Py）	スキル（SQL）
Auto CDC（ストリーミングソース）	CDC フィードから SCD タイプ 1（上書き）またはタイプ 2（履歴）を作成します。	dp.create_auto_cdc_flow()	AUTO CDC INTO ... FROM STREAM	auto-cdc-python	auto-cdc-sql
Auto CDC（定期スナップショット）	連続する完全スナップショットを比較して変更を検出します。	dp.create_auto_cdc_from_snapshot_flow()	AUTO CDC ... FROM SNAPSHOT (DBR 18.3+)	auto-cdc-python	auto-cdc-sql

SCD タイプ 2 の履歴テーブル（__START_AT／__END_AT、特定時点でのクエリ、ファクトと過去のディメンションの結合）のクエリについては、scd-2-querying.md を参照してください。

データ品質 API

機能	説明	Python	SQL	スキル（Py）	スキル（SQL）
Expect（警告）	違反を記録し、すべての行を保持します。	@dp.expect()	CONSTRAINT ... EXPECT (...)	expectations-python	expectations-sql
Expect または削除	違反した行を削除します。	@dp.expect_or_drop()	CONSTRAINT ... EXPECT (...) ON VIOLATION DROP ROW	expectations-python	expectations-sql
Expect または失敗	最初の違反が発生した時点でパイプラインを失敗させます。	@dp.expect_or_fail()	CONSTRAINT ... EXPECT (...) ON VIOLATION FAIL UPDATE	expectations-python	expectations-sql
すべて Expect（警告）	複数の制約を一度に指定し、警告のみを行います。	@dp.expect_all({})	複数の CONSTRAINT 句	expectations-python	expectations-sql
すべて Expect または削除	複数の制約を指定し、違反した場合は削除します。	@dp.expect_all_or_drop({})	DROP ROW を指定した複数の制約	expectations-python	expectations-sql
すべて Expect または失敗	複数の制約を指定し、違反した場合は失敗させます。	@dp.expect_all_or_fail({})	FAIL UPDATE を指定した複数の制約	expectations-python	expectations-sql

データ読み取り API

機能	説明	Python	SQL	スキル（Py）	スキル（SQL）
バッチ読み取り（パイプラインデータセット）	兄弟テーブルを静的 DataFrame として読み取ります。	spark.read.table("name")	SELECT ... FROM name	—	—
ストリーミング読み取り（パイプラインデータセット）	兄弟テーブルをストリーミング DataFrame として読み取ります。	spark.readStream.table("name")	SELECT ... FROM STREAM(name)	—	—
Auto Loader（クラウドファイル）	クラウドストレージから新しいファイルを増分的に取り込みます。	spark.readStream.format("cloudFiles")	STREAM read_files(...)	auto-loader-python	auto-loader-sql
Kafka ソース	Kafka トピックからストリーミング読み取りを行います。	spark.readStream.format("kafka")	STREAM read_kafka(...)	kafka	kafka
Kinesis ソース	AWS Kinesis からストリーミング読み取りを行います。	spark.readStream.format("kinesis")	STREAM read_kinesis(...)	—	—
Pub/Sub ソース	GCP Pub/Sub からストリーミング読み取りを行います。	spark.readStream.format("pubsub")	STREAM read_pubsub(...)	—	—
Pulsar ソース	Apache Pulsar からストリーミング読み取りを行います。	spark.readStream.format("pulsar")	STREAM read_pulsar(...)	—	—
Event Hubs ソース	Azure Event Hubs（Kafka プロトコル）からストリーミング読み取りを行います。	spark.readStream.format("kafka") + EH config	STREAM read_kafka(...) + EH config	kafka	kafka
JDBC / Lakehouse Federation	フェデレーション経由で外部システムからバッチ読み取りを行います。	spark.read.format("postgresql") etc.	フェデレーションカタログ経由でテーブルを直接参照します	—	—
カスタムデータソース	ユーザー定義の Python データソースです。	spark.read[Stream].format("custom")	N/A — Python only	—	—
静的ファイル読み取り（バッチ）	ファイルを一度だけ読み込みます（増分追跡なし）。	spark.read.format("json"|"csv"|...).load()	read_files(...) (no STREAM)	—	—
アップストリームの変更コミットをスキップ	アップストリームテーブル上の CDC コミットを無視します。	.option("skipChangeCommits", "true")	STREAM name WITH (SKIPCHANGECOMMITS)	streaming-table-python	streaming-table-sql

テーブル／スキーマ機能 API

機能	説明	Python	SQL	スキル（Py）	スキル（SQL）
リキッドクラスタリング	複数列にわたる適応型のデータレイアウトです。PARTITION + Z-ORDER を置き換えます。可能な場合は Auto clustering を優先してください	cluster_by=[...]	CLUSTER BY (col1, col2)	materialized-view-python	materialized-view-sql
自動リキッドクラスタリング	Databricks がクエリパターンに基づいてクラスタリングキーを選択します。	cluster_by_auto=True	CLUSTER BY AUTO	materialized-view-python	materialized-view-sql
パーティション列	従来の固定パーティション方式です。リキッドクラスタリングを優先してください。	partition_cols=[...]	PARTITIONED BY (col1, col2)	materialized-view-python	materialized-view-sql
テーブルプロパティ	Delta テーブルのプロパティ（自動最適化、CDF、保持期間）です。	table_properties={...}	TBLPROPERTIES (...)	materialized-view-python	materialized-view-sql
明示的なスキーマ	事前に列の型を宣言します（推論する場合との比較）。	schema="col1 TYPE, ..."	(col1 TYPE, ...) AS	materialized-view-python	materialized-view-sql
生成列	書き込み時に他の列から計算される列です。	schema="..., col TYPE GENERATED ALWAYS AS (expr)"	col TYPE GENERATED ALWAYS AS (expr)	materialized-view-python	materialized-view-sql
行フィルター（パブリックプレビュー）	UC のきめ細かなアクセス制御です。関数で行をフィルタリングします。	row_filter="ROW FILTER fn ON (col)"	WITH ROW FILTER fn ON (col)	materialized-view-python	materialized-view-sql
列マスク（パブリックプレビュー）	UC のきめ細かなアクセス制御です。関数で列をマスクします。	schema="..., col TYPE MASK fn USING COLUMNS (col2)"	col TYPE MASK fn USING COLUMNS (col2)	materialized-view-python	materialized-view-sql
プライベートデータセット	UC に公開されない、マテリアライズされた中間データです。	private=True	CREATE PRIVATE ...	materialized-view-python	materialized-view-sql

レガシー DLT 構文 — 必ず移行してください

上記の表には最新の API のみが記載されています。ユーザーコードに次のいずれかがあれば、レガシー DLT 構文です。必ず最新の形式に移行し、拡張しないでください。変更を提案する前に、変換が正しく行われるよう references/dlt-migration.md をお読みください（特に、apply_changes → create_auto_cdc_flow のセマンティクスと partition_cols → cluster_by に注意してください）。

次の記述があれば…	…DLT です。次の形式に移行してください
import dlt	from pyspark import pipelines as dp
@dlt.table(...)、@dlt.append_flow(...)、@dlt.expect*	dp.* の同じ名前のデコレーター（例：@dp.table、@dp.expect_or_drop）。
@dlt.view(...)（または古いコードにある場合は @dp.view(...)）	@dp.temporary_view(...) — 最新の API に view デコレーターはなく、temporary_view のみです。
dlt.read("name") / dlt.read_stream("name")	spark.read.table("name") / spark.readStream.table("name")
dp.read(...) / dp.read_stream(...)	こちらもレガシーです — spark.read.table(...) / spark.readStream.table(...) を使用してください。
dlt.apply_changes(...) / dp.apply_changes(...)	dp.create_auto_cdc_flow(...)。sequence_by は列名（文字列）または col(...) を受け取ります。stored_as_scd_type は、タイプ 2 には整数の 2、タイプ 1 には文字列の "1" を指定します。
dlt.apply_changes_from_snapshot(...)	dp.create_auto_cdc_from_snapshot_flow(...)
dlt.create_streaming_table(...)	dp.create_streaming_table(...)
SQL の LIVE.<name> プレフィックス	プレフィックスのない名前（バッチ処理では SELECT FROM name、ストリーミング処理では SELECT FROM STREAM(name)）。最新のパイプラインでは LIVE. はエラーになります。
CREATE LIVE TABLE / CREATE STREAMING LIVE TABLE	CREATE OR REFRESH MATERIALIZED VIEW / CREATE OR REFRESH STREAMING TABLE。
CREATE TEMPORARY LIVE VIEW（別名 CREATE LIVE VIEW）	CREATE TEMPORARY VIEW。例外：CREATE TEMPORARY VIEW は、期待値の CONSTRAINT 句をサポートしません。一時ビューで期待値が必要となるまれなケースでは、CREATE LIVE VIEW を引き続き使用できます。詳細は temporary-view-sql.md および expectations-sql.md を参照してください。
APPLY CHANGES INTO ... FROM STREAM ...（SQL）	AUTO CDC INTO ... FROM STREAM ...
partition_cols=[...] / PARTITIONED BY (...) + ZORDER	cluster_by=[...] / CLUSTER BY (...)（リキッドクラスタリング）。
input_file_name()	_metadata.file_path（SQL）/ F.col("_metadata.file_path")（Python）。
create_streaming_table / パイプライン設定の target=... パラメーター	schema=...

言語の選択（Python と SQL）

スキャフォールディングを行う前に決定してください。この選択によって、テンプレートファイル（.py と .sql）と適用するリファレンスドキュメントが決まります。両方を併用することもできますが、主に使う言語を選んでください。迷った場合は、シンプルな SQL をデフォルトにしてください。

ユーザーの要望	選択
「Python パイプライン」、UDF、pandas、ML 推論、pyspark	Python
「SQL パイプライン」、「SQL ファイル」	SQL
「シンプルなパイプライン」、「テーブルを作成」、「集計」	SQL（よりシンプルなため、デフォルトとして使用）
複雑なパラメーター化ロジック、カスタム UDF、ML	Python

曖昧な場合は質問してください。ユーザーが明示的に切り替えない限り、選択した言語を使用してください。

ワークフローの選択

プロジェクトの形態は 3 種類あります。スキャフォールディングの前に選択してください。本番環境向けの作業では A、調査やデモ用のスキャフォールディングでは C をデフォルトにしてください。

• A：新規のスタンドアロンパイプラインプロジェクト（DAB） — パイプライン自体がプロジェクトで、既存の databricks.yml はありません。databricks pipelines init --output-dir . --config-file init-config.json でスキャフォールディングします。→ 1-project-initialization-with-dab.md
• B：既存のバンドル内のパイプライン（DAB） — すでに databricks.yml があります。src/ を参照する resources/<name>.pipeline.yml を追加します。→ 1-project-initialization-with-dab.md#workflow-b-pipeline-in-existing-bundle
• C：CLI による迅速な反復（バンドルなし） — プロトタイピング向けです。databricks pipelines create / start-update / list-pipeline-events を使用します。作業を本番環境に移行する場合は、後からバンドルにまとめます。→ 2-rapid-iteration-with-cli.md

パイプラインの構成

• ユーザーから別の指示がない限り、メダリオンパターン（Bronze → Silver → Gold）に従ってください。デフォルトではシンプルにし、テーブルは数個にとどめます。
• 1 ファイルにつき 1 データセットとし、ファイル名にはデータセット名を使用します。変換ファイルは src/ または transformations/ に配置します。
• Gold レイヤー：重要なビジネスディメンションを保持してください。 Gold に集計する際は、アナリストがフィルターや切り口として使うディメンション（場所、部門、製品ライン、顧客セグメント、期間）を残します。過度に集計すると、下流で復元できない情報が失われます。ダッシュボードについて言及があった場合、そのすべてのフィルター項目を Gold テーブルの列に含める必要があります。失われたディメンションを復元するより、クエリでさらに集計するほうが簡単です。

パイプラインの実行

適切な実行コマンドは、上で選択したワークフローによって異なります。

• ワークフロー A / B（DAB） — コードの変更を反映するには、databricks bundle deploy の実行が必要です。実行、ドライラン、選択的リフレッシュの前には、必ずデプロイしてください。

  databricks bundle validate --profile <profile>
  databricks bundle deploy -t dev --profile <profile>
  databricks bundle run <pipeline_name> -t dev --profile <profile>
  databricks pipelines get <pipeline_id> --profile <profile>      # status
  

  → DAB の実行と反復に関する詳細：references/1-project-initialization-with-dab.md#running-a-pipeline-workflow-a--b

• ワークフロー C（CLI、バンドルなし） — ファイルをワークスペースにアップロードしてから、パイプラインを直接実行します。コードを変更するたびに再アップロードしてください。

  databricks workspace import-dir ./my_pipeline /Workspace/Users/<user>/my_pipeline --overwrite
  databricks pipelines start-update <pipeline_id>
  

  → CLI の実行とポーリングのパターン：references/2-rapid-iteration-with-cli.md

リフレッシュモード（両方のワークフロー）：

• 1 つのテーブルだけを実行する場合は、選択的リフレッシュを推奨します。依存先はすでにマテリアライズされている必要があります。
• フルリフレッシュは最もコストが高く、危険なオプションであり、データ損失につながる可能性があります（ストリーミングソースを最初から再処理し、ストリーミング状態を破棄します）。本当に必要な場合にのみ使用してください。必ず、ユーザーが明示的に承認する必要のある後続の手順として提案してください。

最上位のパイプライン状態ではなく、必ず更新をポーリングしてください。理由については、2-rapid-iteration-with-cli.md#step-4-start-an-update-and-poll-that-update のポーリングに関する説明を参照してください。同じルールはバンドルの実行にも適用されます。

リファレンス一覧

プロジェクトとライフサイクル：

• 1-project-initialization-with-dab.md — ワークフロー A と B。
• 2-rapid-iteration-with-cli.md — ワークフロー C。開始・更新 + ポーリング + エラー抽出。
• pipeline-configuration.md — 作成・更新用の完全な JSON リファレンス + バリエーションのスニペット + 複数スキーマ + プラットフォームの制約。
• performance.md — Liquid Clustering、状態管理、結合、事前集計、モニタリング。
• dlt-migration.md — DLT → SDP への変換。

横断的なパターン：

• streaming-patterns.md — 重複排除、ウィンドウ集計、遅延データ、レスキューデータの隔離、異常検出、ラグのモニタリング。
• scd-2-querying.md — 現在の状態、特定時点の状態、ファクトと履歴ディメンションの結合。
• kafka.md — Kafka / Event Hubs の取り込み。
• real-time-mode.md — リアルタイムモード（RTM）：1 秒未満の継続的なパイプライン（@dp.update_flow、Kafka シンク、サーバーレスまたはクラシック）。

Auto Loader の形式固有オプション：JSON · CSV · XML · Parquet · Avro · Text · ORC。

データセット、フロー、CDC、期待値、Auto Loader、およびシンクのリファレンスは、上記の API リファレンステーブルに機能と言語ごとに記載されています。