使用Prisma进行数据导出与Stream处理

随着数据量的不断增长，如何高效地处理和导出数据成为了开发中的一个重要议题。Prisma作为一个现代化的数据库工具套件，不仅提供了类型安全的查询构建器，还通过其ORM层简化了数据库的交互。本文将重点介绍如何使用Prisma结合Stream来导出大量数据，并探讨其在实际开发中的应用。

Prisma与Stream的基础

Prisma通过其强大的ORM层，允许开发者以声明性的方式操作数据库。而Node.js中的Stream API则提供了处理流式数据的能力，这对于大量数据的导出尤为重要。

什么是Prisma

Prisma是一个开源的数据库工具套件，它帮助开发者使用声明式的数据访问和ORM。Prisma支持多种数据库，如PostgreSQL、MySQL和SQLite，它通过生成标准的数据库模型来与这些数据库进行交互。

Prisma Logo

什么是Stream

在Node.js中，Stream是一种处理流式数据的接口，可以将数据视为一系列的小块（chunk），而不是一次性加载整个数据集到内存中。这对于处理大量数据，尤其是文件导出等操作非常有用。

Node.js Stream

使用Prisma进行数据导出

数据导出的需求

在许多应用中，我们可能需要将数据库中的数据导出到CSV、Excel或者其他格式的文件中。这通常涉及到大量的数据处理和流式写入。

数据导出的实现

整合Query + Readable stream

使用Prisma时，我们可以通过cursor-based pagination来有效地处理大量数据。这种方式不需要像offset-based pagination那样加载大量数据到内存中。

const fetcher = (cursor) => prisma.model.findMany({
  take: 1000,
  skip: cursor ? 1 : 0,
  cursor: cursor ? { id: cursor } : undefined,
});

转换为Readable stream

通过将数据源转换为Readable stream，我们可以逐块读取数据，而不是一次性将所有数据加载到内存中。

const streamQuery = (fetcher) => {
  let cursor;
  return new Readable({
    objectMode: true,
    async read() {
      const items = await fetcher(cursor);
      if (items.length === 0) {
        this.push(null);
      } else {
        for (const item of items) {
          this.push(item);
        }
        cursor = items[items.length - 1].id;
      }
    },
  });
};

格式化并输出到Excel

使用Transform处理数据

将数据流转换为Excel格式，我们可以使用Transform stream。

const formatter = new Transform({
  objectMode: true,
  transform(chunk, _, callback) {
    callback(null, [chunk.a, chunk.b, chunk.c]);
  },
});

queryStream.pipe(formatter);

使用Writable接收数据

最后，我们使用Writable stream将格式化后的数据写入Excel文件。

import Excel from 'exceljs';

const exportToXlsxStream = (header) => {
  const reader = new PassThrough();
  const workbook = new Excel.stream.xlsx.WorkbookWriter({
    stream: reader,
  });

  const worksheet = workbook.addWorksheet('exported');
  const columns = header.map(h => ({ header: h, key: h }));
  worksheet.columns = columns;

  const writer = new Writable({
    objectMode: true,
    write(chunk, _, callback) {
      worksheet.addRow(chunk).commit();
      callback();
    },
  });

  writer.on('finish', async () => {
    worksheet.commit();
    await workbook.commit();
  });

  return { reader, writer };
};