Deno中高效生成大型CSV文件的流式处理教程

// 模拟生成大量数据
const data: { plz: string; strasse: string }[] = [];
for (let i = 0; i < 1000000; i++) {
  data.push({ plz: '12345', strasse: 'Teststrasse-' + i });
}

// 异步迭代器，逐个生成数据
async function* asyncIterator(data: any[]): AsyncGenerator<any> {
  for (const entry of data) {
    yield entry;
  }
}

// 使用第三方库写入CSV（示例，可能性能不佳）
async function writeCSV(data: any[]): Promise<void> {
  // 假设这里使用了某个社区库的writeCSVObjects方法
  // const file = await Deno.open('./test.csv', { write: true, create: true });
  // await writeCSVObjects(file, asyncIterator(data), { header: ['plz', 'strasse'] });
  // file.close();
  console.log("此为传统方式示意，不推荐用于大规模数据");
}

// writeCSV(data);

import { CsvStringifyStream } from "https://deno.land/std@0.219.0/csv/csv_stringify_stream.ts";
import { readableStreamFromIterable } from "https://deno.land/std@0.219.0/streams/readable_stream_from_iterable.ts";

/**
 * 模拟生成大量数据
 * @param count 生成的数据行数
 * @returns 一个包含指定行数的数组
 */
function generateLargeData(count: number): { id: number; name: string; description: string }[] {
  const data: { id: number; name: string; description: string }[] = [];
  for (let i = 0; i < count; i++) {
    data.push({ id: i + 1, name: `Item ${i + 1}`, description: `Description for item ${i + 1}` });
  }
  return data;
}

/**
 * 高效写入CSV文件
 * @param filename 要写入的文件名
 * @param data 要写入的数据数组
 * @param columns CSV文件的列名数组
 */
async function writeLargeCsvEfficiently(
  filename: string,
  data: { [key: string]: any }[],
  columns: string[],
): Promise<void> {
  console.log(`开始写入 ${filename}，数据量：${data.length} 行...`);
  const startTime = performance.now();

  // 1. 打开文件，获取其可写流
  const file = await Deno.open(filename, { create: true, write: true, truncate: true });

  // 2. 将数据数组转换为可读流
  const readable = readableStreamFromIterable(data);

  // 3. 构建数据处理管道
  await readable
    // 管道传输到 CsvStringifyStream，将对象转换为CSV字符串
    .pipeThrough(new CsvStringifyStream({ columns: columns }))
    // 管道传输到 TextEncoderStream，将CSV字符串编码为字节流（UTF-8）
    .pipeThrough(new TextEncoderStream())
    // 管道传输到文件的可写流，完成写入
    .pipeTo(file.writable);

  // 文件写入完成后，Deno会自动关闭文件句柄，但显式关闭也是一个好习惯
  // file.close(); // pipeTo会自动关闭，此处可省略

  const endTime = performance.now();
  console.log(`CSV文件 ${filename} 写入完成，耗时：${((endTime - startTime) / 1000).toFixed(2)} 秒`);
}

// 调用示例
const numberOfRecords = 1_000_000; // 一百万行数据
const largeDataset = generateLargeData(numberOfRecords);
const csvColumns = ["id", "name", "description"];

await writeLargeCsvEfficiently("output_data.csv", largeDataset, csvColumns);

// 尝试写入一个更小的文件以验证
// await writeLargeCsvEfficiently("small_output.csv", generateLargeData(100), csvColumns);