正常性チェックに基づいて Lakehouse テーブルを最適化する

適用対象:✅ Microsoft Fabric の SQL 分析エンドポイント

このチュートリアルでは、インテリジェントテーブルメンテナンスを実行するMicrosoft Fabricパイプラインを構築する方法について説明します。

このソリューションは、Lakehouse SQL 分析エンドポイントで sys.sp_get_table_health_metrics T-SQL ストアドプロシージャを呼び出し、結果を評価し、テーブルに実際にメンテナンスが必要な場合にのみ OPTIMIZE 実行します。この「チェックしてから実行する」パターンにより、健全なテーブルに対する不要なコンピューティングコストを防ぎつつ、劣化したテーブルは自動的にメンテナンスされるようになります。

メンテナンスが必要な理由

Lakehouse テーブルには、時間の経過と同時に小さな Parquet ファイルが多すぎるため、SQL 分析エンドポイントでのクエリパフォーマンスが低下する可能性があります。

このパイプラインは、テーブルの状態に関係なく、固定スケジュールで OPTIMIZE を実行するのではなく、情報に基づいた決定を行います。テーブルの正常性が最初にチェックされ、異常が検出された場合にのみ最適化がトリガーされます。

前提条件

開始する前に、以下の項目があることを確認します:

共同作成者以上のアクセス許可を持つ Microsoft Fabric ワークスペース。
監視する Delta テーブルが少なくとも 1 つ含まれる、そのワークスペース内の Lakehouse。このチュートリアルでは、 SalesDataLakehouseという名前の Lakehouse を使用します。
Fabric データパイプラインに関する知識。
Fabric notebooksに精通していること。

ソリューションの構造

完成したパイプラインには、次の構造があります。

スクリプトアクティビティ: ターゲットテーブルに対して sp_get_table_health_metrics を実行し、テーブルの正常性メトリックを構造化された出力として返します。
条件アクティビティの場合: スクリプト出力から PotentialAnomalyType を直接読み取り、0 より大きいかどうかを確認します。 PotentialAnomalyTypeの詳細については、「潜在的な異常の種類のコード」を参照してください。
ノートブックアクティビティ ( True ブランチ内): Spark ノートブックからテーブルに対して OPTIMIZE を実行します。

このチュートリアルの最後には、パイプラインからパラメーターを受け取り、トリガーされたときにテーブルを最適化するノートブックが用意されています。

手順 1: 最適化ノートブックを作成する

ノートブックは、ターゲットの Lakehouse、スキーマ、テーブル名をパイプラインのパラメーターとして受け入れ、Spark SQL を使用して OPTIMIZE 実行します。

Fabric ワークスペースで、[+ 新しいアイテム>Notebook] を選択します。
ノートブックに Optimize-Table という名前を付けます。
[ 場所] で、確認するテーブルが格納されている Lakehouse を選択します。この演習では、 SalesDataLakehouseという名前の Lakehouse を使用します。
を選択してを作成します。

パラメーターセルを追加する

最初のセルは、パイプラインが実行時にオーバーライドする変数を定義します。

最初のセルに、次のパラメーターを入力します。値は重要ではなく、実行時にパイプラインによってオーバーライドされます。
```
# Parameters 
lakehouse_name = "<LakehouseName>"
schema_name    = "<SchemaName>"
table_name     = "<TableName>"
```
Important

Fabric ノートブックでのパラメーター化のしくみ: 実行時に、Fabricは、パイプラインによって渡された値でこれらの変数を再割り当てするパラメーターセルの直後に新しいセルを挿入します。ここで設定する値は変数を初期化するだけで、読みやすさが向上します。
セルメニュー (...) を選択 >パラメーターセルを切り替えて 、このセルをパラメーターセルとしてマークします。

OPTIMIZE セルを追加する

OPTIMIZE コマンドは、T-SQL コマンドではなく Spark SQL コマンドです。ノートブック、Spark ジョブ定義、Lakehouse メンテナンスインターフェイスなどの Spark 環境で実行する必要があります。 SQL 分析エンドポイントと Warehouse SQL クエリエディターでは、このコマンドは直接サポートされていません。

2 番目のセルに、次のように入力します。

full_name = f"{lakehouse_name}.{schema_name}.{table_name}"
print(f"Optimizing {full_name} ...")

result = spark.sql(f"OPTIMIZE {full_name}")
result.show(truncate=False)

必要に応じて Markdown セルを追加して、他のユーザーのノートブックを適切に文書化します。完成したノートブックは次のようになります。

Note

この例では、スキーマが有効になっている Lakehouse について考えます。 Lakehouse スキーマを使用しない場合は、 full_name の 3 部構成の名前を適宜調整します。

手順 2: パイプラインを作成する

Fabric ワークスペースで、[+ 新しい項目>Pipeline] を選択します。
パイプラインに Check-and-Optimize-Table という名前を付けます。
パイプラインキャンバスの背景を選択し、[ パラメーター] タブを開きます。次の 3 つのパラメーターを追加します。

Name タイプ既定値

lakehouse_name String SalesDataLakehouse

schema_name String dbo

table_name String FactSales

Name	タイプ	既定値
`lakehouse_name`	String	`SalesDataLakehouse`
`schema_name`	String	`dbo`
`table_name`	String	`FactSales`

手順 3: スクリプトアクティビティを追加する

スクリプトアクティビティは、SQL 分析エンドポイントで sys.sp_get_table_health_metrics 実行され、結果をキャプチャします。

Important

ストアドプロシージャ アクティビティではなく、スクリプト アクティビティを使用します。スクリプトアクティビティのみが、ダウンストリームアクティビティが解析できる構造化 JSON 出力として結果セットを公開します。

[ アクティビティ ] タブで、[ スクリプト ] を選択してキャンバスに追加します。
テーブルの正常性を確認する名前を付けます。
[ 設定] タブで、次の手順を実行します。
- 接続: Lakehouse の SQL 分析エンドポイントを選択します。一覧にない場合は、ドロップダウンリストの下部にある [ すべて参照 ] を選択し、Lakehouse の SQL 分析エンドポイントを見つけます。
- スクリプトの種類: [クエリ] を選択します。
- スクリプト: [ 動的コンテンツの追加] を選択し、次の式を入力します。
```
@concat('EXEC sys.sp_get_table_health_metrics ''',
        pipeline().parameters.schema_name, '.',
        pipeline().parameters.table_name, '''')
```

この式は、ターゲットテーブルに対してストアドプロシージャを実行する SQL コマンドを生成します (例: EXEC sys.sp_get_table_health_metrics 'dbo.FactSales')。

スクリプトの出力を確認する

パイプラインを 1 回実行し、 スクリプトアクティビティ の出力を調べます。次のような JSON オブジェクトが表示されます。

{
  "resultSetCount": 1,
  "resultSets": [
    {
      "rowCount": 1,
      "rows": [
        {
          "PotentialAnomalyType": 3,
          "PotentialAnomalyDescription": "Too many small files...",
          "FileCount": 2688,
          "...": "..."
        }
      ]
    }
  ]
}

Important

実際の結果は、テーブルの状態によって異なる場合があります。キーは、 sys.sp_get_table_health_metricsによって公開された列を返すということです。

手順 4: If 条件アクティビティを追加する

If 条件アクティビティは、PotentialAnomalyType アクティビティの出力からを直接読み取り、その結果に基づいて決定を行います。次の手順に従います。

[ アクティビティ ] タブで [ 条件の場合 ] を選択して、キャンバスにアクティビティを追加します。
「異常をチェックする」という名前 を付けます。
[テーブルの正常性の確認] から [成功] (緑色) の矢印を描画して異常を確認します。
If 条件アクティビティの [アクティビティ] タブで、式を次の値に設定します。
```
@greater(int(activity('Check Table Health').output.resultSets[0].rows[0]['PotentialAnomalyType']), 0)
```

この式は、 sys.sp_get_table_health_metricsによって返された最初の行を読み取り、 PotentialAnomalyType を整数にキャストし、値が 0 より大きい場合に true に評価されます。これは、ターゲットテーブルで検出された異常を示します。

手順 5: Notebook アクティビティを追加する (True ブランチ)

[条件の場合] アクティビティが選択されている状態で、[True] の横にある [編集] (鉛筆アイコン) を選択します。キャンバスは、True 分岐をスコープとするサブキャンバスに切り替わります。

ノートブック アクティビティを True サブキャンバスにドラッグします。
名前をRun OPTIMIZEにします。
[設定] タブで、次のようにします。
- ノートブック: 手順 1 で作成した Optimize-Table ノートブックを選択します。
- [ 基本パラメーター] を展開し、次の 3 つの行を追加します。
  
  Name タイプ Value
  
  lakehouse_name String @pipeline().parameters.lakehouse_name
  
  schema_name String @pipeline().parameters.schema_name
  
  table_name String @pipeline().parameters.table_name

Name	タイプ	Value
`lakehouse_name`	String	`@pipeline().parameters.lakehouse_name`
`schema_name`	String	`@pipeline().parameters.schema_name`
`table_name`	String	`@pipeline().parameters.table_name`

3 つの名前の列名は、ノートブックのパラメーターセル内の変数名と 正確に一致する必要があります。

Note

誤ったアクティビティは空欄のままにできます。 If Condition アクティビティは、空の False 分岐を何もしない処理として扱い、パイプラインを成功として報告します。

完成したパイプラインは次のようになります。

手順 6: 検証して実行する

パイプラインツールバーの [検証 ] を選択して、構成エラーを確認します。
[ 実行 ] を選択してパイプラインを手動で実行します。
実行を監視し、確認します。
1. テーブルの正常性を確認する: 実行時に、このアクティビティからの出力を調べます。 sys.sp_get_table_health_metrics ストアドプロシージャからの出力が JSON 形式で表示されます。
2. 異常を確認する: スクリプト出力から直接 PotentialAnomalyType を読み取り、正しく評価します。
3. OPTIMIZE を実行 する ( PotentialAnomalyType > 0場合のみ): 異常チェック アクティビティが True と評価された場合は、 OPTIMIZE の実行 アクティビティの入力を確認して、正しいパラメーター (Lakehouse 名、スキーマ、テーブル名) が使用されていることを確認し、出力を確認して OPTIMIZE 操作からのメッセージを確認します。

リソースをクリーンアップする

このチュートリアル専用のリソースを作成し、不要になった場合は、ワークスペースから次の項目を削除します。

Check-and-Optimize-Tableパイプライン。
Optimize-Table ノートブック。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-06-24