背景
Microsoft Fabric では、Notebook の パラメーターセル を利用することで、Notebook 本体を都度作り替えなくても、入力値や処理対象を柔軟に切り替えられます。
そのため、用途ごとに似たような Notebook を複数作成するのではなく、1 つのパラメーター化された Notebook を再利用しながら運用しやすくなります。
また、Fabric Pipeline の Notebook アクティビティ から ベースパラメーター を使って値を渡すことで、同じ Notebook をさまざまな処理パターンに使い回せます。
今回のシナリオでは、別 Workspace の Lakehouse Files に保存された複数の CSV を、Notebook で読み込み、別 Workspace の Lakehouse Tables に Delta テーブルとして保存する流れを確認します。
参考記事
https://learn.microsoft.com/ja-jp/fabric/data-engineering/author-execute-notebook
https://learn.microsoft.com/ja-jp/fabric/data-factory/parameters
https://learn.microsoft.com/en-us/fabric/data-factory/notebook-activity
https://learn.microsoft.com/ja-jp/fabric/data-engineering/lakehouse-notebook-explore
https://learn.microsoft.com/en-us/fabric/data-engineering/how-to-use-notebook
https://spark.apache.org/docs/latest/sql-data-sources-csv.html
https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.csv.html
今回のシナリオ
コピー元
Workspace1 - Lakehouse1 - Files - 00.Departments- フォルダ内に複数 CSV が保存されている
コピー先
Workspace2 - Lakehouse2 - Tables
やりたいこと
- Notebook で複数 CSV を読み込む
- 別 Workspace の Lakehouse Tables に Delta テーブルとして保存する
- Pipeline からパラメーターを渡して、Notebook の処理内容を動的に切り替える
今回の例では、以下の値をパラメーターとして扱います。
| パラメーター名 | 内容 | 型 |
|---|---|---|
source_folder_abfs |
コピー元フォルダの ABFS/ABFSS パス | String |
target_table_name |
保存先テーブル名 | String |
write_mode |
overwrite / append
|
String |
header |
CSV の 1 行目を列名として扱うか | Bool |
infer_schema |
CSV の型を自動推定するか | Bool |
run_id |
Pipeline 実行 ID | String |
pipeline_name |
Pipeline 名 | String |
Notebook 側のパラメーターは、int、float、bool、string などの 単純型 をサポートします。
list や dict のような複合型はそのままでは渡せないため、必要な場合は JSON 文字列 にして渡し、Notebook 内で逆シリアル化します。
パラメーターが渡る流れ
今回のポイントは次の 3 段階です。
- Pipeline の パラメーター を作る
- Notebook アクティビティ の ベースパラメーター に割り当てる
- Notebook のパラメーターセル で受け取る
Notebook 側では、パラメーターセルの変数名 と、Pipeline 側の ベースパラメーター の Name を一致させる必要があります。
事前に押さえておくポイント
1. コピー先の Lakehouse を既定の Lakehouse にする
Notebook には複数の Lakehouse を追加できます。
そのうち、ピン留めされた Lakehouse が既定の Lakehouse として使われます。
今回のように saveAsTable() でテーブルを書き込む場合は、コピー先の Lakehouse2 を既定にしておくと分かりやすいです。
また、既定の Lakehouse を切り替えた後はセッションの再起動が必要です。
2. Pipeline 実行時の権限に注意する
Notebook を Pipeline Activity として実行する場合、Notebook は Notebook の所有者 や Pipeline の所有者 ではなく、最後に Pipeline を更新したユーザーの ID で実行されます。
クロス Workspace の読み取りや書き込みがある場合は、この権限差で失敗することがあるため注意が必要です。
手順
1. Notebook に Lakehouse を追加する
Notebook の左側から [データ項目の追加] を選択し、Lakehouse を追加します。
- コピー先の Lakehouse2 を追加し、既定の Lakehouse にする
- コピー元の Lakehouse1 も追加する
2. コピー元フォルダの ABFS/ABFSS パスを確認する
コピー元の Files/00.Departments フォルダのパスを確認します。
例:
abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/<lakehouse-id>/Files/00.Departments
Notebook から別 Lakehouse の Files を読む場合は、ABFS/ABFSS の絶対パスを使うと分かりやすいです。
3. Notebook の先頭にパラメーターセルを作成する
Notebook の先頭付近にセルを 1 つ作成し、セル右上の ... メニューから [パラメーター セルを切り替える] を選択します。
パラメーターセル になったことを確認
そして、以下のように、Notebook の先頭にパラメーターセルを作成します。
# =========================================
# パラメーターセル
# このセルを「パラメーター セル」に設定する
# =========================================
source_folder_abfs = ""
target_table_name = ""
write_mode = "overwrite" # overwrite / append
header = True
infer_schema = True
# 任意: 監査用
run_id = ""
pipeline_name = ""
ポイント
-
write_modeは空文字ではなく、最初から"overwrite"などの既定値を入れておく -
headerやinfer_schemaはBool型なのでTrue/Falseを入れる - 型に合った既定値を入れておくと、Notebook 単体実行でも確認しやすい
4. Notebook の処理本体を作成する
パラメーターセルの下に、処理本体を書きます。
from pyspark.sql import functions as F
print(f"source_folder_abfs = {source_folder_abfs}")
print(f"target_table_name = {target_table_name}")
print(f"write_mode = {write_mode}")
print(f"header = {header}")
print(f"infer_schema = {infer_schema}")
print(f"run_id = {run_id}")
print(f"pipeline_name = {pipeline_name}")
if not source_folder_abfs:
raise ValueError("source_folder_abfs が未設定です")
if not target_table_name:
raise ValueError("target_table_name が未設定です")
df = (
spark.read
.format("csv")
.option("header", str(header).lower())
.option("inferSchema", str(infer_schema).lower())
.load(f"{source_folder_abfs}/*.csv")
.withColumn("_pipeline_run_id", F.lit(run_id))
.withColumn("_pipeline_name", F.lit(pipeline_name))
)
(
df.write
.mode(write_mode)
.format("delta")
.saveAsTable(target_table_name)
)
print(f"Saved to table: {target_table_name}")
5. header と infer_schema の意味を理解しておく
header
header=True は、CSV の 1 行目を列名として扱う設定です。
False の場合は、1 行目もデータとして扱われます。
infer_schema
inferSchema=True は、Spark が CSV の内容を見て列型を自動推定する設定です。
数値列を int や double として推定してくれるため、PoC や初期検証では便利です。
ただし、inferSchema を有効にすると、スキーマ推定のために追加で入力を走査するため、大量ファイルや定期実行ではコストが増えることがあります。
本番運用では、必要に応じて inferSchema=False にして schema() を明示的に指定する方法も検討します。
6. Notebook で気を付けること
同じ変数を後続セルで再定義しない
ここは特にハマりやすいポイントです。
Fabric の実行エンジンは、Pipeline から渡された値を パラメーターセルの下に追加された入力セル で上書きします。
そのため、その後のセルで同じ変数を再代入すると、受け取った値を自分で潰してしまいます。
例えば、次のようなセルを後ろに置くのは避けます。
# NG例
source_folder_abfs = ""
target_table_name = ""
write_mode = ""
実際にハマりやすいのは、「Pipeline から値を受け取れていない」のではなく、「Notebook 側で再定義して上書きしている」 ケースです。
パラメーターセルは 1 つにして、同じ変数名は後段で再定義しないのが安全です。
7. Pipeline を新規作成する
Fabric で Pipeline を新規作成します。
8. Pipeline パラメーター を作成する
Pipeline 全体の パラメーター を作成します。
パラメーター は、外部値を受け取って Pipeline を再利用するための仕組みで、一度セットされると実行中には変化しません。
今回の例では、次のように作成します。
-
p_source_folder_abfs: String -
p_target_table_name: String -
p_write_mode: String -
p_header: Bool -
p_infer_schema: Bool
必要に応じて既定値を入れておきます。
9. Notebook アクティビティ を追加する
Pipeline に Notebook アクティビティ を追加し、設定 タブで対象の Notebook を選択します。
10. ベースパラメーター を設定する
Notebook アクティビティ の 設定 タブで、ベースパラメーター を以下のように設定します。
| Name | Type | Value |
|---|---|---|
source_folder_abfs |
String | @pipeline().parameters.p_source_folder_abfs |
target_table_name |
String | @pipeline().parameters.p_target_table_name |
write_mode |
String | @pipeline().parameters.p_write_mode |
header |
Bool | @pipeline().parameters.p_header |
infer_schema |
Bool | @pipeline().parameters.p_infer_schema |
run_id |
String | @pipeline().RunId |
pipeline_name |
String | @pipeline().Pipeline |
ここで重要なのは、Notebook 側の変数名と ベースパラメーター の Name を一致させることです。
ベース パラメーターの設定例：
ここで、動的なコンテンツの追加 を設定する
11. 手動実行時に Pipeline パラメーターを指定する
Pipeline を手動実行するときは、作成した Pipeline Parameters に値を入力してから実行します。
これにより、同じ Pipeline / Notebook を使いながら、フォルダやテーブル名だけを差し替えて再利用できます。
12. 実行確認
Pipeline を実行すると、まずは各アクティビティが順次実行されていることを確認できます。
Notebook アクティビティ が 処理中 になっていれば、Pipeline から Notebook への呼び出しが開始されている状態です。
Pipeline が処理中の状況
次に、手動実行時に指定した Pipeline パラメーター と、実際に Notebook に渡された値が一致しているかを確認します。
特に、source_folder_abfs、target_table_name、write_mode などの値が想定通りに反映されているかを見ておくと安心です。
念のため、渡されたパラメーターと実際の入力値が一致しているかを確認する
Notebook アクティビティ が正常終了すると、Pipeline 上でも 成功 と表示されます。
この状態になれば、Pipeline から Notebook が正常に実行されたことを確認できます。
成功したことを確認する
続いて、実行済みの Notebook Activity 名をクリックし、Notebook スナップショット を確認します。
スナップショットでは、実行時に渡されたパラメーターや、Notebook 内の出力ログを確認できます。
実行されたアクティビティ名をクリックし、Notebook スナップショットを確認する
Notebook スナップショットを開くと、print() で出力した値や、Notebook 内で実際に処理された内容を確認できます。
ここで、Pipeline から受け取った値で CSV 読み込みと saveAsTable() が実行されていることを確認します。
実行された内容を確認する
最後に、コピー先の Lakehouse を開き、Tables 配下に対象テーブルが作成されていることを確認します。
ここまで確認できれば、Pipeline パラメーター → Notebook ベースパラメーター → Notebook パラメーターセル の連携で、別 Workspace の CSV を Lakehouse テーブルとして保存できたことになります。
コピー先の Lakehouse にテーブルが作成されたことを確認する
よくあるハマりどころ
1. パラメーターセルに設定していない
セルを作っただけでは不十分で、[パラメーター セルを切り替える] を実行しておく必要があります。
2. Notebook 側と Pipeline 側で名前が違う
例えば Notebook 側が target_table_name なのに、ベースパラメーター 側が table_name だと値は渡りません。
名前は完全一致 させます。
3. Notebook 側で同じ変数を後続セルで再定義している
一番多いパターンです。
受け取った後に再代入すると、自分で値を上書きしてしまいます。
4. 既定 Lakehouse を変更したのにセッション再起動していない
既定 Lakehouse を切り替えた後は、Notebook セッションの再起動が必要です。
5. 権限不足
クロス Workspace の読み書きがある場合は、Pipeline を最後に更新したユーザー の権限も確認しておきます。
まとめ
Microsoft Fabric で Notebook を Pipeline から動的に実行する場合は、次の流れを押さえておくと整理しやすいです。
- Notebook に パラメーターセル を作る
- Pipeline に パラメーター を作る
- Notebook Activity の ベースパラメーター で渡す
- Notebook 側で値を受け取って処理する
- 後続セルで同じ変数を再定義しない
この形にしておくと、Notebook を大量に複製せずに、1 つの汎用 Notebook を再利用できます。
特に、フォルダ名・テーブル名・書き込みモードなどが変わるだけの処理では、かなり運用しやすくなります。
参考記事
https://learn.microsoft.com/ja-jp/fabric/data-engineering/author-execute-notebook
https://learn.microsoft.com/ja-jp/fabric/data-factory/parameters
https://learn.microsoft.com/en-us/fabric/data-factory/notebook-activity
https://learn.microsoft.com/ja-jp/fabric/data-engineering/lakehouse-notebook-explore
https://learn.microsoft.com/en-us/fabric/data-engineering/how-to-use-notebook
https://spark.apache.org/docs/latest/sql-data-sources-csv.html
https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.csv.html
Disclaimer
※本記事の一部は AI を活用して下書き・表現整理を行っています。内容については公開前に筆者自身で確認・修正していますが、製品仕様や画面表示、機能名は将来的に変更される可能性があります。実際の検証や利用にあたっては、必要に応じて Microsoft Learn などの最新の公式情報もあわせてご確認ください。