e2e-medallion-architecture

テレメトリ — 必須。 すべての api.fabric.microsoft.com 呼び出しには、 x-ms-fabric-skill: e2e-medallion-architecture を含める必要があります（az rest の場合は --headers "x-ms-fabric-skill=e2e-medallion-architecture"）。 すべての LRO ポーリング、fabric_lro、リトライにも適用されます。スニペットでは省略されていますが、必ず追加してください。

重要な注意事項

1. ワークスペース名からワークスペースの詳細（ID を含む）を確認するには、すべてのワークスペースを一覧表示し、JMESPath でフィルタリングします
2. ワークスペース ID、アイテムの種類、アイテム名からアイテムの詳細（ID を含む）を確認するには、そのワークスペース内の該当する種類のアイテムをすべて一覧表示し、JMESPath でフィルタリングします

エンドツーエンドのメダリオンアーキテクチャ

前提知識

このスキルが依存する基礎知識について、以下の関連ドキュメントをお読みください。

• COMMON-CORE.md — Fabric REST API のパターン、認証、トークンのオーディエンス、アイテムの検出
• COMMON-CLI.md — az rest、az login、トークンの取得、CLI 経由の Fabric REST
• SPARK-AUTHORING-CORE.md — ノートブックのデプロイ、レイクハウスの作成、ジョブの実行
• notebook-api-operations.md — ノートブックの作成に必須 — .ipynb の構造要件、セル形式、getDefinition／updateDefinition のワークフロー

Spark 固有の最適化の詳細については、data-engineering-patterns.md を参照してください。

---

アーキテクチャの概要

メダリオンアーキテクチャは、3 つの段階的なレイヤーで構成されるデータレイクハウスのパターンです。

レイヤー	目的	最適化プロファイル	ユースケース
Bronze（Raw）	受信したままの生データを取り込む	書き込み最適化、追記のみ、取り込み日でパーティション分割	監査証跡、再処理、リネージ
Silver（Cleaned）	重複を除去し、検証および調整したデータ	読み書きのバランスを最適化、業務日でパーティション分割	特徴量エンジニアリング、運用レポート
Gold（Aggregated）	分析用の事前計算済みメトリック	読み取り最適化（ZORDER、コンパクション）、月／年でパーティション分割	Power BI レポート、ダッシュボード、SQL エンドポイント経由のアドホック分析

• Bronze：スキーマオンリード — 柔軟なスキーマ。Delta のタイムトラベルで監査とロールバックに対応
• Silver：スキーマ強制 — 準拠しない書き込みを拒否します。ソースが変更された場合は、mergeSchema でスキーマの進化に対応します
• Gold：厳格なスキーマガバナンス — 厳選され、業務部門の承認を受けたデータセットのみ

---

必須／推奨／回避

必須事項

• infrastructure-orchestration.mdを参照し、スキーマ対応の可否に基づいてレイクハウスアーキテクチャを選択します。
  • **推奨：**スキーマ対応レイクハウス → bronze、silver、gold スキーマを持つワークスペースを1つ、レイクハウスを1つ作成します
  • **レガシー：**スキーマ非対応 → ガバナンスとアクセス制御のため、レイヤーごと（Bronze、Silver、Gold）に個別のワークスペースを作成します
• スキーマとテーブルの作成には Livy API を使用します。スキーマ対応レイクハウスでスキーマとテーブルを作成するには、Livy セッション経由で Spark SQL ステートメントを送信します（POST /livyApi/versions/2023-12-01/sessions → POST .../statements）。これは、Fabric レイクハウスで DDL 操作（CREATE SCHEMA、CREATE TABLE）を行う唯一のプログラムによる REST 経路です。
• Bronze にメタデータ列を追加します。取り込みタイムスタンプ、ソースファイル、バッチ ID を含めます
• Bronze から Silver への変換にデータ品質ルールを適用します（重複排除、NULL の処理、範囲検証）
• メダリオンアーキテクチャのすべてのレイヤーのテーブルに Delta Lake 形式を使用します
• Silver／Gold への書き込みでは、変更のないデータの再処理を避けるため、パーティションを考慮した上書きを使用します
• 各レイヤーの後に検証ステップを含めます（行数、スキーマチェック、異常検知）
• REST API 経由でノートブックを作成する際は、notebook-api-operations.mdの**.ipynb 検証と Fabric 固有の注意点**に従います。すべてのコードセルに"outputs": []と"execution_count": nullを含める必要があります
• エンドツーエンドのフローを最後まで完了します。ノートブックの作成後に止めず、ユーザーが部分的なセットアップを明示的に要求しない限り、必ずレイクハウスをバインドし、ノートブックを順番に実行（Bronze → Silver → Gold）し、結果を検証して、Power BI を Gold レイヤーに接続します
• MLV とノートブックの比較を推奨する際は、MLV にはスキーマ対応レイクハウスが必要であることを明記します。MLV の定義と増分更新のレビューは spark-cli のauthoringモードに引き継ぎます。スケジュール設定、更新、監視、障害診断は spark-cli のmlvモードに引き継ぎます
• MLV の定期更新では、正確な操作手順 Lakehouse → Materialized lake views → Manage → Schedules を案内します。自動化には POST /workspaces/{workspaceId}/lakehouses/{lakehouseId}/jobs/refreshMaterializedLakeViews/schedules を使用します。/mlvRefreshSchedules を作り上げたり、定期更新をノートブックのスケジュール設定経由にしたりしてはいけません

推奨事項

• 全件更新ではなく、増分処理（ウォーターマークパターン）を優先します
• 個別にテストおよびデバッグできるよう、レイヤーごとにノートブックを分けます
• Gold テーブルでは、頻繁にフィルターされる列に ZORDER を適用します
• Silver 層と Gold 層では、書き込み後に OPTIMIZE を実行します
• 環境に応じた Spark 設定（Bronze は書き込み重視、Silver はバランス重視、Gold は読み取り重視）を使用します
• 重複を避けてコンシューマーのワークスペースに Gold データを公開するには、OneLake ショートカットを使用します
• 各レイヤーの所有者を明確にします。エンジニアは Bronze／Silver、アナリストは Gold を担当します
• レイヤー間のパスと設定を一元管理するには、Fabric Variable Libraries を使用します
• 中程度または高いガバナンス要件がある場合は、複数ワークスペースへのデプロイパターンを使用します（Bronze／Silver／Gold を別々のワークスペースに配置）
• 変換を Spark SQL で表現でき、宣言的な更新セマンティクスの利点がある場合は、Silver／Gold テーブルに Materialized Lake Views（MLV）を使用します。spark-cli — Materialized Lake View patternsとMLV incremental refresh patternsを参照してください
• 「materialized view」、「spark materialized view」、「MLV」は、同じ Fabric 機能を指すものとして扱います

避けるべきこと

• スキーマを使わずにすべてのレイヤーを単一のレイクハウスに保存すること — スキーマを使わないレイクハウスでは、RLS／CLS と MLV を有効にするために、ノートブックの初期化セルまたは Environment の構成で OneLake Spark Catalog を有効にする必要があります。スキーマを利用できない場合は、分離のためにレイクハウスを分けてください。
• スキーマ対応のレイクハウスを利用できるのに、レイクハウスを3つ作成すること — 代わりに、1つのレイクハウス内でスキーマを使用してください（よりシンプルで、初期化セルの定型コードが不要になり、MLV のスキーマ間変換も効率的になります）。
• Silver レイヤーを省略して、Bronze から直接 Gold に進むこと
• ワークスペース ID、レイクハウス ID、FQDN をハードコードすること — REST API で取得してください
• Bronze テーブルに対して、LIMIT を指定せずに SELECT * を実行すること（テーブルは際限なく大きくなります）
• 下流の依存関係を確認せずに VACUUM を実行すること
• メダリオンレイヤー間（Bronze→Silver→Gold）で OneLake ショートカットを連鎖させること — 系譜とガバナンスのため、各レイヤーは物理的に実体化する必要があります
• 実装コード全体をスキルにコピーすること — 代わりに、LLM が生成するように誘導してください
• 外部の HTTP／HTTPS URL から Spark で直接読み込むこと — Fabric Spark は任意の外部 URL にアクセスできません。まず curl、OneLake API、または Fabric パイプラインの Copy アクティビティを使ってレイクハウスの Files/ にデータを配置し、その後レイクハウスのパスから読み込んでください
• .ipynb の構造を検証せずに REST API でノートブックを作成すること — コードセルに execution_count: null または outputs: [] がないと、エラーが表示されずに失敗したり、「Job instance failed without detail error」が発生したりします

---

ワークスペースのセットアップに関するガイダンス

メダリオンワークスペースをセットアップする際は、まずアーキテクチャパターンを選択してください（詳細なガイダンスは infrastructure-orchestration.md を参照してください）。

オプション A：スキーマ対応レイクハウス（推奨）

1. 単一のワークスペースを作成する：{project}-{env}
2. スキーマを含む 1つのレイクハウスを作成する：{project}_lakehouse
3. レイクハウス内にスキーマを作成する：
   • 生データの取り込み用に bronze スキーマを作成する
   • クリーニング・検証済みデータ用に silver スキーマを作成する
   • 集計分析用に gold スキーマを作成する
4. 変換方法を選択する：
   • オプション 4a： 各レイヤーでノートブックを使用する（PySpark または Spark SQL による変換）
   • オプション 4b： 増分更新に対応した宣言型変換には、マテリアライズドレイクビュー（Spark SQL）を使用する（クエリが IR の対象となる場合）。materialized-lake-view-patterns.md および mlv-incremental-refresh-patterns.md を参照してください
   • 注： PySpark MLV は利用できますが、完全更新のみ（増分更新なし）です。UDF／複雑な Python ロジックが必要な場合に使用してください
   • MLV の利点： スキーマ対応レイクハウスでは OneLake Spark Catalog が 自動的に有効になる ため、MLV はノートブックの初期化セルや Environment の構成なしですぐに使用できます
5. RBAC（任意）：スキーマ内で行レベルのセキュリティと列のマスキングを使用して、きめ細かなアクセス制御を実現します（OneLake Spark Catalog も必要です）

オプション B：Lakehouse を分離する（レガシー）

1. 3 つのワークスペースを作成します：
   • {project}-bronze-{env}
   • {project}-silver-{env}
   • {project}-gold-{env}
2. 各ワークスペースに 1 つずつ Lakehouse を作成します：
   • Bronze ワークスペース → {project}_bronze Lakehouse
   • Silver ワークスペース → {project}_silver Lakehouse
   • Gold ワークスペース → {project}_gold Lakehouse
3. レイヤーごとのワークスペースに RBAC を割り当てます：
   • Bronze：取り込み／エンジニアリングの書き込み権限
   • Silver：エンジニアリング／データ品質の権限
   • Gold：より厳格なキュレーション制御を適用した、分析／BI 利用者のアクセス権
4. スキーマレスの Lakehouse で OneLake Spark Catalog を有効にします（RLS／CLS およびカタログに基づくアクセスパターンで必須です）：
   • 推奨： Environment で spark.sql.fabric.catalog.enable-schemaless-lakehouses=true を設定し、ノートブックにアタッチします。
   • 代替方法： ノートブックから既定の Lakehouse バインドを省略します。4 部構成の完全修飾参照（workspace.lakehouse.schema.table）を使用します。既定の Lakehouse が設定されていない場合、OneLake Spark Catalog は自動的に有効になります。
   • 代替方法（内部向け／非サポート）： すべてのノートブックの最初のセルに次を追加します：

   %%pyspark
   !echo "spark.sql.fabric.catalog.enable-schemaless-lakehouses=true" >> /home/trusted-service-user/.trident-context
   

   • ⚠️ 注： この回避策では、今後の Fabric リリースで変更される可能性がある内部ランタイム構成パスを使用します。安定した、文書化された OneLake Spark Catalog のサポートを得るには、スキーマ対応の Lakehouse を推奨します。
   • この構成では、スキーマレスの Lakehouse で次がサポートされます：
     • ✅ 行レベルセキュリティ（RLS）と列レベルセキュリティ（CLS）
   • 注： MLV にはスキーマ対応の Lakehouse（オプション A）が必要です。スキーマレスの Lakehouse では、Delta テーブルを使用するノートブックを使用してください。

共通の手順（両方のオプション）

上記のオプション A またはオプション B を完了した後、次の手順を実行します：

1. 各レイヤーのノートブックを作成します（変換ステージごとに 1 つ）。.ipynb の検証と Fabric 固有の注意事項に従ってください。
2. 各ノートブックを対応する Lakehouse にバインドします。正しい Lakehouse ID を metadata.dependencies.lakehouse に設定します（notebook-api-operations.md § Default Lakehouse Binding を参照）：
   • オプション A：すべてのノートブックで同じ Lakehouse を使用し、スキーマプレフィックス（bronze.table、silver.table）を使用します。
   • オプション B：
     • Bronze ノートブック → Bronze ワークスペース／Lakehouse
     • Silver ノートブック → Silver ワークスペース／Lakehouse（クロスワークスペース OneLake アクセス／完全修飾参照を介して Bronze を読み取ります）
     • Gold ノートブック → Gold ワークスペース／Lakehouse（クロスワークスペースアクセスを介して Silver を読み取ります）
3. ノートブックのデプロイを確認します。updateDefinition が Succeeded を返したことを確認してください。これで、コンテンツと Lakehouse バインドが保持されたことを確認できます。再確認のために getDefinition を呼び出してはいけません。これは非同期 LRO であり、不要な遅延が加わります。
4. ノートブックを順番に実行します。まず Bronze、次に Silver、最後に Gold の順に実行します。実行構成で正しい defaultLakehouse を指定した POST .../jobs/instances?jobType=RunNotebook を使用します（id と name の両方が必須です）。
5. Power BI を Gold レイヤーに接続します。Gold Lakehouse の SQL エンドポイントを特定し、Direct Lake セマンティックモデルを作成して、Gold サマリーテーブルのビジュアルを含むレポートを作成します（Gold Layer → Power BI Consumption を参照）。
6. Bronze → Silver → Gold のフローを繰り返し実行するためのオーケストレーション用パイプラインを作成します。

明示的なオーバーライド：単一ワークスペース

ユーザーが単一ワークスペースへのデプロイを明示的に求めた場合（たとえば、POC／小規模チーム／モノリシックパターン）は、現在の方法を維持します：

• 1 つのワークスペースに Bronze／Silver／Gold の Lakehouse を分けて配置します。
• ワークスペースを共有する場合も、論理的なレイヤー分離を維持します。
• 複数ワークスペース設計と比較したガバナンス上のトレードオフを明示します。

環境ごとにパラメーターを設定します：ワークスペース名のサフィックス（-dev、-prod）、データ量（サンプル／全量）、容量 SKU、Bronze の保持期間。

---

Bronze レイヤー — 取り込みパターン

ユーザーから Bronze レイヤーへのデータ取り込みを依頼された場合、LLM には次のように案内してください。

1. まずデータをレイクハウスに配置する：Spark が読み取る前に、外部データをレイクハウスの Files/ フォルダーにステージングする必要があります。次のいずれかを使用してください。
   • Fabric Pipeline Copy アクティビティ（定期的な読み込みに推奨） — 外部ソース（HTTP、FTP、データベース、クラウドストレージ）に接続し、OneLake に書き込みます
   • OneLake API / curl — storage.azure.com トークンを使用して REST API 経由でファイルをアップロードします（COMMON-CLI.md § OneLake Data Access を参照）
   • OneLake Shortcut — Azure ADLS Gen2、S3、または別の OneLake の場所にデータがすでにある場合に使用します
   • notebookutils.fs — ノートブック内でマウント済みストレージパスからコピーします
   • ⚠️ Fabric Spark は任意の HTTP/HTTPS URL から読み取れません — spark.read.format("csv").load("https://...") は失敗します
2. レイクハウスのパスから読み取る：データが Files/ に格納されたら、レイクハウス相対パスを使用して読み取ります（例：spark.read.format("csv").load("Files/landing/daily/")）
3. メタデータを追加して書き込む：追跡用の列（取り込みタイムスタンプ、ソースファイル、バッチ ID）を追加し、わかりやすい名前の Delta テーブルを作成して、取り込み日でパーティション分割し、追加モードで書き込みます
4. 検証する：行数をログに記録し、スキーマ構造を検証して、過去のパターンと比べて異常があればフラグを立てます

---

Silver レイヤー — 変換パターン

ユーザーから Bronze から Silver への変換を依頼された場合、LLM には次のように案内してください。

• 品質ルール：自然キーまたは複合キーで重複を排除し、無効な範囲を除外し、NULL を処理します（必須項目は削除、任意項目は補完）。論理的な制約を検証します
• スキーマへの準拠：列名を snake_case に統一し、データ型を標準化して、派生列（期間、割合、カテゴリ）を追加します
• スキーマの進化：ソーススキーマが変更された場合は mergeSchema オプションを使用します。Gold テーブルおよび Power BI データセットの下流更新を調整します
• 書き込み戦略：業務日でパーティション分割し、パーティションを考慮した上書きを行い、書き込み後に OPTIMIZE を実行して、書き込み前後のメトリクスをログに記録します

---

Gold レイヤー — 集計パターン

ユーザーから Gold 分析テーブルの作成を依頼された場合、LLM には次の生成を案内してください。

• 一般的な集計：日次／週次／月次のサマリー、ディメンション分析（場所、カテゴリ、タイプ別）、時間経過による傾向の内訳、需要パターン（時間帯、曜日）
• Spark セッション構成 — 書き込み処理を始める前に、Gold ノートブックで次のプロパティを設定してください。

  spark.conf.set("spark.sql.parquet.vorder.default", "true")
  spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
  spark.conf.set("spark.databricks.delta.optimizeWrite.binSize", "1g")
  

  • V-Order（vorder.default）— Fabric の列指向ソート最適化をすべての Parquet ファイルに適用し、Direct Lake と SQL エンドポイントの読み取りパフォーマンスを大幅に向上させます
  • Optimize Write（optimizeWrite.enabled）— 小さなパーティションを最適なサイズのファイル（binSize あたり約 1 GB を目標）に統合し、ファイル数を減らしてスキャン効率を高めます
• 最適化：フィルター列に ZORDER を適用し、書き込み後に OPTIMIZE を実行して、実行時の計算を避けるためにメトリクスを事前集計します

---

エンドツーエンドの実行フロー

メダリオンアーキテクチャをエンドツーエンドで設定する場合、LLM はノートブックの作成とコードのデプロイ後に処理を止めてはいけません。完全なライフサイクルは次のとおりです。

Create Resources → Deploy Content → Bind Lakehouses → Execute → Verify Results

手順

1. レイヤーごとのワークスペースとレイクハウスを作成する（デフォルト） — 各レイヤー（Bronze、Silver、Gold）にワークスペースとレイクハウスを1つずつ作成し、ワークスペース ID とレイクハウス ID を記録します
2. ノートブックを作成する — 各レイヤーに1つずつ、有効な .ipynb 構造のノートブックを作成します（notebook-api-operations.md を参照）
3. 各ノートブックにレイクハウスをバインドする — .ipynb ペイロードに metadata.dependencies.lakehouse を含め、次の項目を指定します。
   • default_lakehouse：対象のレイクハウス GUID
   • default_lakehouse_name：レイクハウスの表示名
   • default_lakehouse_workspace_id：ワークスペース GUID
4. ノートブックの内容をデプロイする — Base64 エンコードした .ipynb ペイロード（内容とレイクハウスのバインドを一緒に含む）を使用して updateDefinition を実行します
5. デプロイを確認する — 各 updateDefinition LRO が Succeeded を返したことを確認します。それで十分です。再確認のために getDefinition を呼び出さないでください。これは非同期 LRO であり、ノートブックごとに大幅な遅延が発生します。
6. ノートブックを順番に実行する — POST .../jobs/instances?jobType=RunNotebook を使用します。
   • executionData.configuration で id と name の両方を指定して defaultLakehouse を渡します
   • 最初に Bronze を実行し、Completed になるまでポーリングしてから Silver を実行し、ポーリングし、続けて Gold を実行してポーリングします
   • 送信前に最近のジョブを確認します（重複を防止します。SPARK-AUTHORING-CORE.md を参照）
7. 結果を確認する — 各ノートブックの完了後、想定されるテーブルが存在し、行数が妥当であることを確認します
8. Power BI を Gold に接続する — Gold のサマリーテーブルにセマンティックモデルとレポートを作成します（Gold レイヤー → Power BI での利用 を参照）

よくある失敗：ノートブックの作成後に処理を止める

バインドや実行を行わずにノートブックコードのデプロイ後にフローを止めると、次の問題が発生します。

• ノートブックにレイクハウスのコンテキストがないため、実行時に spark.sql() と相対パス（Tables/、Files/）が失敗します
• アーキテクチャは設定されてもテストされず、ユーザーには出力や結果が表示されません
• ユーザーから特定の手順で停止するよう明示的に求められない限り、必ず手順 7 まで完了してください

---

Gold レイヤー → Power BI での利用

Gold テーブルへのデータ取り込み後、Power BI を接続して分析結果を表示します。
DirectLake を使用して、Gold レイクハウス上にセマンティックモデルを構築します。

手順

1. Gold レイクハウスの SQL エンドポイントを検出する — GET /v1/workspaces/{workspaceId}/lakehouses/{goldLakehouseId} を呼び出し、properties.sqlEndpointProperties.connectionString と provisioningStatus を取得します。ステータスが Success になるまで待機します。
2. SQL 経由で Gold テーブルを確認する — sqlcmd を使用して SQL エンドポイントに接続し（COMMON-CLI.md § SQL / TDS Data-Plane Access を参照）、対象のテーブルが存在することを確認します。

   SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'nyc_taxi_daily_summary'
   

3. セマンティックモデルを作成する — セマンティックモデルの作成と TMDL のデプロイには、semantic-model-authoring スキルを使用します。POST /v1/workspaces/{workspaceId}/items を type: "SemanticModel" で呼び出して作成し、TMDL 形式を使用して updateDefinition で定義をデプロイします（ITEM-DEFINITIONS-CORE.md § SemanticModel を参照）。
   • モデルのデータソースとして、Gold レイクハウスの SQL エンドポイントを参照する必要があります。
   • Gold のサマリーテーブル（例：nyc_taxi_daily_summary）に対応するテーブルマッピングを定義します。
   • Direct Lake モードを使用します。データをインポートせずに OneLake 内の Delta テーブルに直接接続できます。
   • 興味深い主要な集計値（例：Total Trips、Avg Fare、Total Revenue、Month over Month Growth）のメジャーを含めます。
4. Power BI レポートを作成する — POST /v1/workspaces/{workspaceId}/items を type: "Report" で呼び出して作成し、PBIR 形式を使用して updateDefinition で定義をデプロイします（ITEM-DEFINITIONS-CORE.md § Report を参照）。
   • definition.pbir を通じて、手順 3 で作成したセマンティックモデルを参照します。
   • Gold のサマリーテーブルを使用したビジュアルを含むページを 1 つ以上定義します。
   • 推奨するビジュアル：折れ線グラフ（日次推移）、カード（KPI の合計値）、棒グラフ（カテゴリ別）、テーブル（詳細表示）
5. エンドツーエンドで検証する — セマンティックモデルに対して DAX クエリを実行し、メタデータを検出するには semantic-model-authoring スキルを使用します。Gold テーブルからレポートまでデータが流れることを確認します。

原則

• SQL エンドポイントを動的に検出する — 接続文字列はレイクハウスの応答にある properties.sqlEndpointProperties.connectionString に含まれています。ハードコードしないでください。
• SQL エンドポイントのプロビジョニングを待つ — 接続する前にステータスが Success である必要があります。新しく作成したレイクハウスのプロビジョニングには数分かかる場合があります。
• Direct Lake モードを優先する — データの重複を避けられます。セマンティックモデルは OneLake の Delta テーブルから直接読み取ります。
• テーブル名と列名を正確に一致させる — セマンティックモデルのテーブル定義には、Gold レイクハウスにある Delta テーブル名と列名を正確に使用する必要があります。
• セマンティックモデルの作成（TMDL、更新、アクセス許可）については、semantic-model-authoring スキルを参照します。
• DAX クエリの検証については、メタデータの検出と検証用の DAX クエリに関して、semantic-model-authoring スキルを参照します。

---

パイプラインのオーケストレーション

ユーザーがメダリオンフロー用のパイプラインをリクエストした場合は、次の方針で設計するよう LLM に案内します。

• 構成：順次実行するアクティビティ（Bronze → Silver → Gold）を設定し、それぞれ前の処理の成功を待機します。独立した Gold の集計処理は並列実行できるようにし、検証および通知のアクティビティを含めます。
• パラメーター化：パイプラインレベルの処理日付（デフォルトは昨日）を設定し、すべてのノートブックに渡します。動的な日付式を使用します。
• スケジュール：ソースの更新に合わせて毎日実行し、ウォーターマークベースの増分処理と、修正に備えた定期的な全件更新を行います。
• エラー処理：一時的な障害にはバックオフ付きで再試行し、障害が継続する場合はアラートを送信します。上流で障害が発生した場合は、下流で前回のデータを使用するなど、適切に縮退させます。

---

環境の最適化

Spark の詳細な構成と最適化戦略については、data-engineering-patterns.md を参照してください。

レイヤー	プロファイル	主な設定
Bronze	書き込み負荷が高い	V-Order を無効化し、autoCompact を有効化、大きなファイルサイズを目標に設定、ingestion_date でパーティション分割
Silver	バランス型	V-Order を有効化、アダプティブクエリ実行、業務日付でパーティション分割、フィルター対象の列に ZORDER を適用
Gold	読み取り負荷が高い	V-Order（spark.sql.parquet.vorder.default=true）、Optimize Write（optimizeWrite.enabled=true、binSize=1g）、ベクトル化リーダー、アダプティブ実行、すべてのフィルター対象列に ZORDER を適用、メトリクスを事前集計

---

例

例 1：メダリオンワークスペースをセットアップする（デフォルト）

プロンプト：「売上分析用に、Bronze、Silver、Gold のワークスペースを分けたメダリオンアーキテクチャをセットアップしてください」

LLM が生成するもの：次の REST API 呼び出し：

1. ワークスペースを作成：sales-bronze-dev、sales-silver-dev、sales-gold-dev
2. 各ワークスペースにレイクハウスを 1 つ作成：sales_bronze、sales_silver、sales_gold
3. ワークスペース／レイヤーごとに RBAC ロールを割り当て

# Workspace creation (see COMMON-CLI.md for full patterns)
cat > /tmp/body.json << 'EOF'
{"displayName": "sales-analytics-dev"}
EOF
workspace_id=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "https://api.fabric.microsoft.com/v1/workspaces" \
  --body @/tmp/body.json --query "id" --output tsv)

# Create Bronze lakehouse
cat > /tmp/body.json << 'EOF'
{"displayName": "sales_bronze", "type": "Lakehouse"}
EOF
az rest --method post --resource "https://api.fabric.microsoft.com" \
  --url "https://api.fabric.microsoft.com/v1/workspaces/$workspace_id/items" \
  --body @/tmp/body.json

例 2：Bronze の取り込みを設計する

プロンプト：「メタデータ列を追加して、日次の CSV ファイルを Bronze レイクハウスに取り込んでください」

LLM が生成するもの：次の処理を行う PySpark ノートブック：

1. スキーマ推論または明示的なスキーマを使用してソースファイルを読み込む
2. ingestion_timestamp、source_file、batch_id 列を追加する
3. 取り込み日付でパーティション分割した Delta テーブルに書き込む
4. 行数と検証メトリクスを記録する

# Bronze ingestion pattern (guide LLM to generate full implementation)
from pyspark.sql.functions import current_timestamp, input_file_name, lit
import uuid

batch_id = str(uuid.uuid4())
df = (spark.read.format("csv").option("header", True).load("/Files/landing/daily/")
      .withColumn("ingestion_timestamp", current_timestamp())
      .withColumn("source_file", input_file_name())
      .withColumn("batch_id", lit(batch_id)))
df.write.mode("append").partitionBy("ingestion_date").format("delta").saveAsTable("bronze.events_raw")

例 3：Bronze から Silver への変換

プロンプト：「Bronze データをクリーンアップしてください。重複を削除し、無効なレコードをフィルターし、派生列を追加して、Silver に書き込んでください」

LLM が生成するもの：品質ルールとスキーマ整合性を適用し、最適化を行ってパーティション分割して書き込む PySpark ノートブック。

例 4：エンドツーエンドのパイプライン

プロンプト：「毎日午前 2 時に Bronze の取り込み、続いて Silver の変換、最後に Gold の集計を実行するパイプラインを作成してください」

LLM が生成するもの：ノートブックアクティビティを順次実行する、日付パラメーター、再試行ロジック、スケジュールトリガーを含むパイプライン JSON 定義。