0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@fahmifahim

Microsoft Fabric Lakehouse のデータを CSV として出力する方法（Copy Activity / Notebook）

0
Posted at

Microsoft Fabric の Lakehouse に保存しているデータを CSV として出力する方法について、備忘録としてまとめます。

大量データを CSV として取得したい場合、主に次のような方法が考えられます。

  • Fabric Data Factory の Copy Activity を利用して CSV を生成する
  • Fabric Notebook（Spark）を利用して CSV を生成する
  • 生成したファイルを OneLake File Explorer から取得する
  • 大容量ファイルの転送では AzCopy を利用する

この記事では、特に Copy Activity と Notebook を利用する方法を紹介します。

1. Fabric Data Factory の Copy Activity で CSV を出力する

シンプルに Lakehouse のテーブルを CSV に出力したい場合は、Fabric Data Factory の Copy Activity を利用する方法が分かりやすいです。

構成イメージは以下です。

Lakehouse Table
    ↓
Fabric Data Pipeline
    ↓
Copy Activity
    ↓
Lakehouse Files / CSV

コピー元を指定する

Fabric Pipeline の Copy Activity を追加し、コピー元として Lakehouse のテーブルを指定します。

image.png

コピー先を CSV にする

コピー先には Lakehouse を指定し、保存先を Files、ファイル形式を DelimitedText として設定します。

image.png

CSV の出力では、例えば以下の設定が利用できます。

  • ファイル拡張子（.csv など）
  • 先頭行をヘッダーとして出力
  • 文字コード
  • 区切り文字
  • ファイルあたりの最大行数
  • ファイル名プレフィックス

特に大量データの場合は、1つの大きな CSV にまとめるのではなく、ファイルあたりの最大行数を指定して複数ファイルへ分割する方法も選択できます。

詳細なオプションについては、Microsoft Learn の区切りテキスト形式のドキュメントを確認してください。

補足：File name prefix の制約

ファイルあたりの最大行数 とあわせて ファイル名プレフィックス を指定できますが、コピー元がファイルベースのストアの場合など、一部の構成では利用できません。

その場合、以下のようなエラーが発生することがあります。

FileNamePrefixNotSupportFileBasedSource

この場合は、File name prefix の設定を削除し、自動生成されるファイル名を利用する方法が簡単です。

Pipeline を実行する

設定が完了したら Pipeline を実行します。

実行が正常に完了したことを確認します。

image.png

実行後、対象 Lakehouse の Files フォルダーを確認すると CSV が出力されています。

image.png

あとは必要に応じて CSV をダウンロードします。

OneLake explorer 利用してダウンロードすることもできます。
image.png

2. Fabric Notebook から CSV を出力する

もう1つの方法として、Fabric Notebook から Spark を利用して CSV を生成できます。

Notebook を利用すると、CSV に出力する前に以下のような処理を追加しやすい点がメリットです。

  • 必要な列だけを抽出する
  • 条件を指定してデータを絞り込む
  • データを加工する
  • 出力先を動的に変更する
  • ファイル分割を制御する

以下は、一定行数ごとにデータを分割して CSV として出力するサンプルです。

あくまで動作確認用のサンプルです。実際のデータ量や要件に合わせて調整してください。

from pyspark.sql import Window
from pyspark.sql.functions import monotonically_increasing_id, row_number, col

# 元テーブル（ハイフンを含むテーブル名なので、バッククォートで囲んで有効な識別子にする）
source_table = "`TableA-products`"

# 出力先のベースパス（この直下にファイルを複数作成する）
output_base_path = "Files/ExportCsv/Notebook/run_date=2026-08-17"

# 1つのCSVファイルあたりの行数を指定
rows_per_file = 10

# テーブル名のバッククォートを含めた文字列を、そのまま識別子として解釈させるために
# spark.read.table() ではなく spark.table() を使用する。
df = spark.table(source_table)

# 行番号を採番する（テーブル全体に対して1,2,3,... の連番を振る）
# ※特定の列で並び順を安定させたい場合は orderBy に対象列を指定してください。
w = Window.orderBy(monotonically_increasing_id())
df_with_rn = df.withColumn("rn", row_number().over(w))

# 行番号に基づいて、どのファイルに出力するかのインデックスを計算する
# 例: rows_per_file = 10 の場合
#   rn=1〜10   -> file_index=0
#   rn=11〜20  -> file_index=1
#   ...
df_with_index = df_with_rn.withColumn(
    "file_index",
    ((col("rn") - 1) / rows_per_file).cast("int")
).drop("rn")

# file_index ごとに分割して、別々のCSVに出力する
file_indices = [row[0] for row in df_with_index.select("file_index").distinct().collect()]

for idx in sorted(file_indices):
    target_path = f"{output_base_path}/products_{idx:05d}.csv"
    (
        df_with_index
        .filter(col("file_index") == idx)
        .drop("file_index")  # 出力CSVには file_index 列を含めない
        .coalesce(1)          # 各グループを1ファイルにまとめる
        .write
        .mode("overwrite")
        .option("header", "true")
        .option("encoding", "UTF-8")
        .csv(target_path)
    )

このコードで行っていること

処理内容は大きく以下のとおりです。

  1. Lakehouse のテーブルを Spark DataFrame として読み込む
  2. データ全体に行番号を付ける
  3. rows_per_file で指定した行数ごとに file_index を割り当てる
  4. file_index ごとにデータを抽出する
  5. 各データを CSV として Files 配下に出力する

例えば rows_per_file = 10 の場合、1〜10行目、11〜20行目、といった単位で分割して出力します。

Spark の CSV 出力で注意する点

Spark の .csv(path) は、一般的な Python のファイル出力とは少し動作が異なります。

例えば、

.csv("Files/ExportCsv/products_00000.csv")

と指定しても、products_00000.csv という単一ファイルが直接作成されるわけではなく、基本的にはその名前のフォルダーが作成され、その配下に part-xxxxx.csv が生成されます。

固定したファイル名が必要な場合は、CSV 出力後に NotebookUtils のファイル操作機能を利用して、part-*.csv を移動・リネームする方法が考えられます。

3. 大量データの場合は、よりシンプルな分割方法もある

上記のコードは、行番号を振ってファイル単位を細かく制御するサンプルです。

一方、数千万行のような大量データでは、全体に行番号を付けたうえで collect() と coalesce(1) を繰り返す方法は処理負荷が高くなる可能性があります。

単純に「1ファイルあたりの最大レコード数を指定して CSV を分割したい」という用途であれば、Spark の maxRecordsPerFile を利用する方がシンプルです。

source_table = "`TableA-products`"
output_path = "Files/ExportCsv/Notebook/run_date=2026-08-17"

df = spark.table(source_table)

(
    df.write
    .mode("overwrite")
    .option("header", "true")
    .option("encoding", "UTF-8")
    .option("maxRecordsPerFile", 1000000)
    .csv(output_path)
)

この例では、1ファイルあたり最大 1,000,000 レコードを目安として複数の CSV に分割します。

ファイル名自体は Spark によって part-xxxxx.csv の形式で生成されるため、業務上決まったファイル名が必要な場合は、出力後にリネーム処理を追加します。

4. 出力した CSV を取得する方法

OneLake File Explorer

OneLake File Explorer を利用すると、Windows Explorer から OneLake 上のファイルへアクセスできます。

そのため、Copy Activity や Notebook で作成した CSV を、Windows 上から確認・取得する方法として利用できます。

Microsoft Learn：
https://learn.microsoft.com/ja-jp/fabric/onelake/onelake-file-explorer

※サンプル画面
image.png

AzCopy

大量のファイルや大容量データをまとめて転送する場合は、AzCopy も選択肢になります。

AzCopy は大規模なデータ移動向けのツールで、OneLake とのデータ転送にも利用できます。

一般ユーザーが手動でファイルを取得する用途というよりは、IT 管理者やデータエンジニアが大量データを転送する場合に向いています。

Microsoft Learn：
https://learn.microsoft.com/ja-jp/fabric/onelake/onelake-azcopy

まとめ

Lakehouse のデータを CSV として出力する方法としては、主に以下の方法が考えられます。

方法 用途
Data Factory Copy Activity シンプルにテーブルを CSV 化したい場合
Fabric Notebook データ加工や細かい出力制御が必要な場合
OneLake File Explorer 生成済み CSV を Windows から取得したい場合
AzCopy 大容量ファイルをまとめて転送したい場合

単純な CSV 出力であれば、まずは Data Factory の Copy Activity が分かりやすいと思います。

一方、データ加工や出力方法を細かく制御したい場合は Notebook が便利です。

基本的には、

Lakehouse
  ↓
Copy Activity / Notebook
  ↓
Lakehouse Files / OneLake
  ↓
CSVを取得

という構成で考えるとシンプルです。

参考

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?