Power BI の生データを CSV/ZIP でストリーミング配信する — Node.js Streams と archiver の地雷処理

第 1 弾で Power BI のレポートを「見せる」仕組みを、第 2 弾で executeQueries からデータを「取り出す」設計を組みました。完結編の本記事は「届ける」— 前回予告した Node.js Streams の backpressure と archiver の地雷原を歩きます。

取り出した行をユーザーに渡す最も素朴な実装は、全テーブルの全行をメモリに貯めてから ZIP を作り res.send する形です。これは数十万行までは動いてしまい、数百 MB 級のテーブルが来た日に OOM で死にます。目指す構成は、行の供給 (ページング) → CSV 化 → ZIP 化 → HTTP 応答を一度もメモリに貯めずに繋ぐストリーミングです。

ただしこの構成には、第 2 弾で扱った「無音のデータ欠落」と同じ病が形を変えて潜んでいます。backpressure の扱いを誤れば行が黙って消え、archiver の扱いを誤ればプロセスごと落ちるか、応答がタイムアウトまで終端しない状態になります。本記事では筆者が実環境で運用しているストリーミング実装を、地雷の位置と対策に沿って解説します。

認証 (Managed Identity) は第 1 弾、ページング設計 (0 行終端・実取得行数 offset・3 箇所のエラー検査) は第 2 弾のとおりで、本記事では再掲しません。同期応答をやめて非同期ジョブ化する話は、運用の章で出口だけ示します。

全体像 — 行の供給からブラウザ到達まで

パイプラインは 4 段です。冒頭の図と同じ番号で、各段に潜む地雷を対応付けておきます。

先に設計原則をひとつ宣言します。HTTP のストリーミング応答は、送り始めた後にステータスコードを変えられません。途中で何かが失敗しても、クライアントには既に 200 が届いています。だから目標は「失敗をゼロにする」ことではなく、失敗したら必ず「壊れた ZIP」という検出可能な形で終わることです。最悪なのは、欠けたデータを持つ正常な ZIP が完成してしまうこと — 第 2 弾から続く「無音の成功を許さない」がここでも軸になります。

ハンドラの外形は次のとおりです。

// 1 ページ目の成功を確認してからヘッダを送り、archiver を応答へ直結する
app.get("/api/export", async (req, res) => {
  const firstPages = await fetchFirstPages(tables); // 失敗すればまだ 4xx/5xx を返せる
  res.set("Content-Type", "application/zip");
  res.set("Content-Disposition", 'attachment; filename="export.zip"');
  const archive = archiver("zip", { store: true });
  archive.pipe(res);
  await appendTables(archive, tables, firstPages);
  await archive.finalize();
});

「1 ページ目を先に取ってからヘッダを送る」のは、権限エラーや設定ミスのような始まる前に分かる失敗を正しい HTTP ステータスで返すためです (躓きポイントの章で詳述)。以降の章で、この骨格の中身を上流から順に作ります。

CSV 化 — 列名の剥がし方と RFC 4180

行キーは 3 形ある — 正規表現で中身だけを取る

executeQueries が返す行の JSON キーは、素の列で 受注明細[金額]、テーブル名に空白等があると '受注 明細'[金額]、DAX の SELECTCOLUMNS で別名を付けると [別名] — の 3 形になります。CSV のヘッダにはブラケットの中身だけを使いたいので、正規表現で剥がします。

// 行キー 3 形 ('T'[C] / T[C] / [C]) から列名だけを剥がす
const COLUMN_KEY_RE = /^(?:'(?:[^']|'')*'|[^[\]]*)\[(.+)\]$/;

function columnName(key) {
  const m = COLUMN_KEY_RE.exec(key);
  return m ? m[1] : key;
}

ポイントは (.+) の貪欲マッチです。列名自体が ] を含みうるため、最後の ] までを列名として取る必要があります。テーブル名側の '...' はクォート内の '' エスケープにも対応させています。日本語のテーブル名・列名もこのままで通ります (実環境で確認済み)。

RFC 4180 のエスケープと CRLF

CSV の方言問題は、RFC 4180 に素直に従うのが結局いちばん互換性が高いです。

// RFC 4180: 引用符・カンマ・改行を含むならクォートし、" は "" にする
function csvField(value) {
  const s = value == null ? "" : String(value);
  return /[",\r\n]/.test(s) ? `"${s.replace(/"/g, '""')}"` : s;
}

const line = (cells) => cells.map(csvField).join(",") + "\r\n";

ルールは 3 つだけ — フィールドに " , 改行を含むならダブルクォートで囲む、フィールド内の " は "" に重ねる、行末は CRLF。pandas / polars / DuckDB / Excel のいずれも素直に読める最大公約数です。

BOM は付けない — ただし「読み手」で分ける

この CSV には UTF-8 BOM を付けません。出力の想定読者が pandas / polars / DuckDB といったデータ処理系で、BOM が付いていると先頭列名が 受注日 のように化けるからです。一方、Excel でダブルクリックして開く前提の CSV は BOM がないと文字化けします。筆者のアプリではこの 2 種類が意図的に混在しており、エスケープ関数は共用しつつ、BOM の付与だけを呼び出し側の責務にしています。RFC 4180 は BOM に言及しないので、ここは規格ではなく読み手のツールで決める領域です。

ストリーミング — backpressure を「検出可能」に扱う

公式の最短経路: Readable.from(asyncGenerator)

まず正攻法から。Node.js でこの種のストリームを作る最短経路は async generator です。

// 公式の最短経路 — yield が消費側に同期するので backpressure は自動
async function* csvChunks(tableName) {
  yield line(headerOf(tableName));
  for await (const rows of fetchPages(tableName)) {
    yield rows.map((r) => line(toCells(r))).join(""); // ページ単位で束ねる
  }
}
const csvStream = Readable.from(csvChunks(table), { objectMode: false });

generator の yield は消費側の需要に同期して停止・再開するため、backpressure はランタイムが自動で正しく処理します。新規に書くならこれで十分なケースがほとんどです。落とし穴をひとつだけ挙げると、Readable.from の objectMode は既定が true なので、バイトストリームとして流すなら false の明示を忘れないことです。

それでも筆者が手書きした理由

筆者の実装は、この generator ではなく Readable の手書きです。「Readable.from では不十分だから」ではありません — 上記のとおり機構としては十分です。手書きを選んだ理由は 2 つあります。

第 2 弾から続くテーマは「無音の失敗を仕組みで検出する」ことでした。backpressure の扱いを誤ると起きるのは、まさに無音の行喪失です。筆者はこの不変条件を言語仕様への信頼として暗黙に置くのではなく、pageIdx というテスト可能な状態として明示したかった — それが手書きの理由です。どちらを選んでも、次の契約を守ることに変わりはありません。

push() が false でも、行は捨てない

手書き Readable の中核です。push() が false を返したら「消費が追いついていないから送るのを止めろ」の合図ですが、このとき手元の行を捨ててはいけません。ページ内のどこまで流したかをカーソル (pageIdx) で覚えておき、次の _read() で続きから再開します。

// push() が false でも行は捨てない — pageIdx が「どこまで流したか」を覚える
const flushPage = () => {
  let buf = "";
  while (pageIdx < pageRows.length) {
    buf += line(toCells(pageRows[pageIdx]));
    pageIdx += 1;
    if (buf.length >= CHUNK_TARGET_BYTES) {
      if (!readable.push(buf)) return false; // 続きは次の _read() から
      buf = "";
    }
  }
  return buf ? readable.push(buf) : true;
};

これは Node.js 公式ドキュメントが実装者向けに示している「push() が false を返したらソースからの読み出しを止め、次に _read() が呼ばれたら再開する」パターンそのものです。加えて _read() は消費のたびに何度も呼ばれるため、ページ取得の pump を running フラグで 1 本に制限し、二重 fetch を防ぎます。

もうひとつ、コード中の CHUNK_TARGET_BYTES (筆者は 64KB) にも意味があります。1 行ずつ push すると、100k 行のページで 100k 回の Buffer 確保と backpressure 判定が走ります。バイトストリームの既定 highWaterMark がちょうど 64KiB なので、その粒度まで束ねてから流すとオーバーヘッドと流量制御のバランスが取れます。

ZIP 化 — archiver の地雷原

CSV ストリームを ZIP に束ねるのは npm の archiver を使います。デファクトのライブラリですが、エラーハンドリングまわりに実環境で踏んだ地雷が集中しています。

store モードで直列に append する

// 無圧縮 (store) で、テーブルごとに entry 完了を待って直列に append する
const archive = archiver("zip", { store: true });
archive.pipe(res);
for (const table of tables) {
  const name = entryName(table);
  const done = waitForEntry(archive, name); // 'entry' イベントを Promise 化
  archive.append(csvStreamOf(table), { name });
  await done;
}
await archive.finalize();

判断は 2 つ。store (無圧縮) にしているのは、CSV の圧縮に CPU を使うより、同期応答ではワーカーを他リクエストに空ける方が価値が高いからです。直列 append にしているのは、同一セマンティックモデルへの並列 executeQueries が容量に厳しく、応答時間もほぼ改善しないからです。

'error' リスナが 0 だとプロセスごと落ちる

archiver は EventEmitter なので、Node.js の仕様として 'error' イベントにリスナが 1 つも無い状態でエラーが起きると、例外が throw されてプロセスが終了します。さらに archive.append(stream) は内部で PassThrough へ pipe しますが、pipe はエラーを転送しません。archive 側にリスナを張っても、CSV ストリーム側の 'error' はリスナ 0 のまま未捕捉例外になります。

// リスナは append より前に、archive と CSV ストリームの両方へ張る
archive.on("error", (err) => log.error("zip stream failed", err));
const csvStream = csvStreamOf(table);
csvStream.on("error", (err) => {
  log.error("csv stream failed", err);
  archive.destroy(err); // 応答を error 終端させる (次節)
});
archive.append(csvStream, { name });

誤解しないでほしいのは、このリスナはプロセスを守るためだけにある点です。エラーが起きた応答は中断され、クライアントには中央ディレクトリが欠けた壊れた ZIP が届きます。unzip が失敗するので、欠けたデータが正常な顔をして通ることはありません — 冒頭の設計原則どおり「検出可能な失敗」です。

中断は abort() では足りない — destroy(err) まで打つ

クライアントがダウンロードを中断したときの後始末にも罠があります。archiver の abort() は公式ドキュメントに「best-effort であり、残りのソースは drain しない」と明記されており、entry の処理が進行中だと内部ストリームが end されません。結果、応答がインフラの request timeout まで終端せずハングします。

// クライアント切断: 待ちを解いてから abort + destroy で確実に終端する
res.on("close", () => {
  if (res.writableFinished) return; // 正常完了の close は無視
  rejectPendingEntry(new Error("client disconnected"));
  archive.abort();                  // 公式 API (ただし best-effort)
  archive.destroy(new Error("aborted"));
});

destroy(err) は archiver のドキュメントには載っていませんが、archiver は Transform ストリームを継承しているため、ストリーム標準の API としてそのまま使えます。docs に無いので見落とされがちですが、これを打って初めて応答が error 終端します。あわせて 'close' を拾うこと自体も重要です。拾わないと、waitForEntry の Promise が永久に pending のまま残り、進行中の fetch ごとリークします。

エントリ名 — 消毒と一意化はセットで

ZIP のエントリ名にはテーブル名を使いますが、パス区切りや制御文字は消毒 (_ に置換) が必要です。そして消毒には二次災害があります: 受注/明細 と 受注_明細 のような別テーブルが消毒後に同名へ潰れると、ZIP に重複エントリができるだけでなく、'entry' イベントを名前で待つ Promise が別テーブルの完了で誤って解決されます。

// 消毒と一意化はセット — 消毒で潰れた同名は _2, _3 で分ける
const used = new Map();
function entryName(tableName) {
  const base = tableName.replace(/[\\/:*?"<>|\x00-\x1f]/g, "_");
  const n = (used.get(base) ?? 0) + 1;
  used.set(base, n);
  return n === 1 ? `${base}.csv` : `${base}_${n}.csv`;
}

なお日本語のエントリ名はそのままで問題ありません。ZIP 仕様の general purpose bit 11 (UTF-8 フラグ) を archiver が立てるため、Windows のエクスプローラーで開いても文字化けしないことを実環境で確認しています。

テスト — fetchImpl を全レイヤで注入する

この一連の実装で、テストの効きを決めたのは設計上のひと工夫でした。Power BI を呼ぶ全レイヤで fetchImpl を引数として注入可能にしてあることです。

// fetchImpl を注入できれば、Power BI なしでページングの不変条件を検証できる
it("切り詰めページの続きを回収する", async () => {
  const fetchImpl = mockResponses([
    pageOf(62_000, { truncated: true }), pageOf(38_000), pageOf(0),
  ]);
  const rows = await collectPages(fetchImpl, "受注明細");
  expect(rows).toHaveLength(100_000);
});

方針はシンプルで、SDK や認証に依存する部分はテストせず、純関数 (csvField・columnName・entryName の一意化) と、fetchImpl のモックで駆動できるサーバロジック (ページングの 0 行終端、切り詰めの回収、push が false を返した後の pageIdx 再開) に厚くテストを置きます。第 2 弾・本記事で積んだ不変条件は、すべてこの形の vitest で固定してあります。「設計の正しさを言葉で説明できるものは、モックでテストに固定できる」— 逆に言うと、注入点のない実装はこの検証ができません。

躓きポイント

応答開始後のエラーは HTTP では伝えられない

ストリーミング応答は 200 を送った瞬間に後戻りできません。途中で executeQueries が失敗しても、ステータスを 500 に変える手段はもうありません。筆者は 2 段構えにしています。第 1 に、最初のページ取得が成功するまでヘッダを送らない — 権限エラー・設定ミス・DAX エラーといった「始まる前に分かる失敗」は正しい 4xx/5xx で返します。第 2 に、送出開始後の失敗は前章の destroy による途中終端 = 壊れた ZIP として受け入れます。クライアント側では unzip が必ず失敗するので、無音では終わりません。

Content-Length は付けられない

ストリーミングなので応答は chunked になり、合計サイズを事前に約束できません。store モードなら概算は可能ですが、概算値を Content-Length に書くとズレた時点でプロトコル違反になります。付けない判断のトレードオフとして、ブラウザのダウンロード進捗バーが「残り時間不明」になります。これは受け入れました — 正確なサイズの約束より、途中終端が「不完全なダウンロード」として素直に検出されることを優先しています。

運用の勘どころ

同期ストリーミングの上限はインフラの request timeout。 第 2 弾で触れたとおり、筆者の環境 (Azure Container Apps) は既定 240 秒で応答が切られます。テーブル数と行数が育ってここに恒常的に触れ始めたら、この記事の構成の卒業時期です — 非同期ジョブでZIP を Blob Storage に書き出し、完了通知と URL 配布へ移行します。

メモリ監視で backpressure の効きを確認する。 正しく効いていれば、エクスポート中の RSS は「1 ページ分 + highWaterMark」程度で頭打ちになります。ダウンロードの遅いクライアントで RSS が右肩上がりになるなら、どこかで push() の戻り値を無視しています。

壊れた ZIP の問い合わせには unzip -t。 本記事の設計では「壊れた ZIP が届く」ことは異常系の正しい姿です。問い合わせを受けたら unzip -t で整合性を確認してもらい、サーバー側の error ログ (切断か、上流の失敗か) と突き合わせるのが最短です。

archiver のバージョン方針を決めておく。 archiver は v7 (CommonJS) と v8 (ESM・Node 18+) で読み込み方が分かれます。どちらを使うかはアプリのモジュール形式に合わせ、後述の経緯も踏まえてリリースを追跡してください。

制約事項 (2026 年 8 月時点)

• archiver 8.0.0 は 2026 年 5 月に約 2 年ぶりにリリースされたものです。休眠期にはフォークが生まれる動きもあったため、メンテナンス状況は継続的な観察をおすすめします
• 本記事の CSV は BOM なし (pandas / polars / DuckDB 前提) です。Excel でダブルクリックして開く用途には BOM 付きの別出力が必要です
• ZIP は無圧縮 (store) です。転送帯域がボトルネックの環境では、CPU との再トレードオフで zlib 圧縮を検討してください
• 同期応答の実行時間はインフラの request timeout で頭打ちになります (Container Apps 既定 240 秒)
• executeQueries 側の制約 (100k 行 / 1M 値 / 15MB / RLS 不可ほか) は第 2 弾を参照してください
• 仕様・ライブラリは更新されていくため、最新の公式情報を確認してください

まとめ — シリーズを閉じる

3 部作を通して作ってきたものを 1 表にまとめます。

貫いてきた軸は「無音の失敗を許さない」です。HTTP 200 の裏の切り詰めは 3 箇所のエラー検査で、backpressure の行喪失は pageIdx の保持で、配信中の失敗は壊れた ZIP という検出可能な形で — 失敗そのものは消せなくても、「失敗が成功の顔をして通り抜ける」ことだけは仕組みで潰せます。

このシリーズの方式が役目を終えるとしたら、Arrow 版 executeDaxQueries が GA し、行数上限とストリーミングの制約が API 側で解決されたときでしょう。その日まで、この 3 本が同じ道を歩く方の地雷除去の役に立てば幸いです。

参考リンク

• Stream (Node.js Documentation) — 手書き Readable の push/false パターンと highWaterMark の仕様
• Backpressuring in Streams (Node.js Learn) — backpressure を無視した場合のメモリへの影響
• Error events (Node.js Documentation) — 'error' リスナ 0 で throw する EventEmitter の仕様
• archiver API (Archiver) — abort() の best-effort 仕様と各イベント
• node-archiver Releases (GitHub) — 8.0.0 (ESM / Node 18+) のリリースノート
• Common Format and MIME Type for CSV Files — RFC 4180 (IETF) — CSV のクォート規則と CRLF
• APPNOTE.TXT - .ZIP File Format Specification (PKWARE) — general purpose bit 11 (UTF-8 ファイル名フラグ)