海量数据导出方案选型：前端 Streams API 与后端异步生成的取舍

// 前端流式拉取并写入 CSV 的核心逻辑
async function exportLargeDataset(apiEndpoint, fileName) {
  // 1. 获取本地文件句柄（Chromium 核心浏览器支持）
  const fileHandle = await window.showSaveFilePicker({
    suggestedName: fileName,
    types: [{
      description: 'CSV File',
      accept: { 'text/csv': ['.csv'] }
    }]
  });
  
  const writableStream = await fileHandle.createWritable();
  
  // 2. 发起分块读取请求（假设后端返回 application/x-ndjson）
  const response = await fetch(apiEndpoint);
  const reader = response.body.getReader();
  const decoder = new TextDecoder('utf-8');
  
  // 3. 写入表头
  await writableStream.write("ID,用户名,交易金额,创建时间\n");

  let buffer = '';
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n');
    buffer = lines.pop(); // 保留未完整的最后一行

    // 4. 将解析出的行格式化并流式写入本地磁盘
    let csvChunk = '';
    for (const line of lines) {
      if (!line.trim()) continue;
      const row = JSON.parse(line);
      csvChunk += `${row.id},${row.username},${row.amount},${row.createdAt}\n`;
    }

    if (csvChunk) {
      await writableStream.write(csvChunk);
    }
  }

  // 处理剩余 buffer 并关闭
  if (buffer.trim()) {
    const row = JSON.parse(buffer);
    await writableStream.write(`${row.id},${row.username},${row.amount},${row.createdAt}\n`);
  }

  await writableStream.close();
}

import { S3Client } from "@aws-sdk/client-s3";
import { Upload } from "@aws-sdk/lib-storage";
import { PassThrough } from "stream";
import { stringify } from "csv-stringify";

async function processExportTask(taskId, dbCursor) {
  const s3 = new S3Client({ region: "ap-northeast-1" });
  
  // 建立一个内存流通道
  const passThrough = new PassThrough();
  const csvStream = stringify({ header: true });

  // 管道连接：CSV 生成器 -> 传输通道
  csvStream.pipe(passThrough);

  // 利用 S3 的 Multipart Upload 机制实现流式写入，不落本地磁盘
  const parallelUploads3 = new Upload({
    client: s3,
    params: {
      Bucket: "company-exports",
      Key: `exports/${taskId}.csv`,
      Body: passThrough,
      ContentType: "text/csv",
    },
  });

  // 模拟从 DB 游标逐行拉取，防止 Worker 内存爆炸
  (async () => {
    for await (const row of dbCursor) {
      csvStream.write([row.id, row.username, row.amount, row.createdAt]);
    }
    csvStream.end();
  })();

  // 等待 S3 分块上传完毕
  await parallelUploads3.done();
  
  // 更新数据库任务状态为 COMPLETED，并记录文件 Key
  await updateTaskStatus(taskId, 'COMPLETED', `exports/${taskId}.csv`);
}