📤 ExportUtils — 高性能 MySQL / MongoDB 数据导出工具

一个支持 流式导出、大表处理、进度追踪 和 自动压缩 的 Node.js 导出工具，适用于将 MySQL 或 MongoDB 数据高效导出为 CSV（GZIP 压缩）。

✨ 特性

✅ 流式导出：内存占用低，支持百万级数据导出
✅ 自动 GZIP 压缩：输出 .csv.gz 文件，节省磁盘与带宽
✅ 实时进度追踪：通过 Redis 存储任务状态（支持查询）
✅ 字段格式化：支持自定义列 formatter（如时间格式转换）
✅ 安全防护：表名/字段名校验（防 SQL 注入）Raw SQL 白名单限制（仅允许简单 SELECT）
✅ 背压控制：自动 pause/resume 流，防止内存溢出
✅ 连接池管理：MySQL 使用连接池，确保高并发安全

🚀 安装

npm install @sch_cat/export-utils

🔧 初始化

import { ExportUtils } from './src/index.js';

const exporter = new ExportUtils({
  redisOptions: {
    host: '127.0.0.1',
    port: 6379,
    // 其他 ioredis 配置...
  },
  mysqlOptions: {
    host: 'localhost',
    user: 'root',
    password: 'password',
    database: 'mydb',
    connectionLimit: 10,
    // 其他 mysql2.PoolOptions...
  },
  mongoOptions: {
    url: 'mongodb://localhost:27017',
    options: {
      // MongoClientOptions
    }
  }
});

💡 如果只使用一种数据库，可省略另一项配置（如仅用 MySQL，则无需 mongoOptions）。

🚀 使用示例

1. 导出 MySQL 表

const taskId = await exporter.createMysqlExportTask({
  table: 'users',
  outputPath: 'users', 
  columns: [
    { key: 'id' },
    { key: 'name', header: '姓名' },
    { key: 'created_at', formatter: (v) => new Date(v).toISOString() }
  ],
  where: "status = 'active'",
  onCompleted: ({ taskId, rowCount, outputPath }) => {
    console.log(`✅ 任务  $ {taskId} 完成，共  $ {rowCount} 行，文件:  $ {outputPath}`);
  },
  onErr: ({ errMsg }) => {
    console.error('❌ 导出失败:', errMsg);
  }
});

console.log('任务ID:', taskId);

2. 使用 Raw SQL

const taskId = await exporter.createMysqlExportTask({
  rawSql: 'SELECT id, email FROM users WHERE created_at > "2024-01-01"',
  outputPath: 'active_users',
  // ...其他配置
});

🔒 rawSql 必须是 简单 SELECT 语句，禁止包含 ;、DROP、UPDATE 等危险操作。

3. 导出 MongoDB 集合

const taskId = await exporter.createMongoExportTask({
  db: 'myapp',
  collection: 'orders',
  outputPath: 'orders.csv',
  query: { status: 'paid' },
  columns: [
    { key: '_id' },
    { key: 'amount', header: '金额' },
    { key: 'createdAt' }
  ],
  onCompleted: ({ outputPath, rowCount }) => {
    console.log(`📦 导出完成:  $ {outputPath} ( $ {rowCount} 行)`);
  }
});

4. 使用 pipeline 聚合管道

const taskId = await exporter.createMongoExportTask({
  db: 'myapp',
  collection: 'orders',
  outputPath: 'orders',
  pipeline: [
    { $match: { time: { $gte: '2024-01-01 00:00:00' } } },
    { $group: { _id: '$storeId', count: { $sum: 1 } } },
  ],
  columns: [
    { key: '_id', header: '门店' },
    { key: 'count', header: '数量' }
  ],
  onCompleted: ({ outputPath, rowCount }) => {
    console.log(`📦 导出完成:  $ {outputPath} ( $ {rowCount} 行)`);
  }
});

💡 由于进度需要先查询总数，在管道聚合查询可能会有性能问题。

5. 查询任务状态

const statuses = await exporter.getTaskStatus({
  taskIdList: [taskId]
});

console.log(statuses[0]);
// 输出示例:
// {
//   taskId: 'export-mysql-csv-xxx',
//   total: 10000,
//   current: 8500,
//   progress: '85.00%',
//   status: 'RUNNING'
// }

💡 进度信息存储在 Redis 中，有效期为 1 小时（TTL = 3600 秒）。

📂 输出文件

• 所有文件默认保存在 ./exports/ 目录（可被 outputPath 覆盖）
• 文件名自动追加 .csv.gz 后缀（如 users → users.csv.gz）
• 支持绝对路径：/data/backups/users.csv

🛠 配置选项

ExportUtilsOptions

| 字段 | 类型 | 必填 | 说明 | |------|------|------|------| | redisOptions | RedisOptions | ✅ | ioredis 连接配置 | | mysqlOptions | PoolOptions | ❌ | mysql2 连接池配置 | | mongoOptions | { url: string; options?: MongoClientOptions } | ❌ | MongoDB 连接配置 |

ExportMysqlOptions / ExportMongoOptions

| 字段 | 说明 | |------|------| | outputPath | 输出路径（必填） | | columns | 指定导出列及格式化函数 | | onCompleted / onErr | 回调钩子 | | MySQL 独有：table, where, rawSql | 三选一（优先 rawSql） | | MongoDB 独有：db, query | |

字段定义（columns）

{
  key: string;        // 数据库字段名（支持嵌套，如 'user.name'）
  header?: string;    // CSV 列标题（默认等于 key）
  formatter?: (value: any) => any; // 值转换函数
}

🧪 示例代码

• 项目中提供了完整示例：
  • examples/export-mysql.ts
  • examples/export-mongo.ts

运行方式：

npm run dev:mysql
npm run dev:mongo

📜 开源协议

MIT © [email protected]

⚠️ 注意事项

1. Raw SQL 安全限制 rawSql 必须满足：
   • 以 SELECT 开头
   • 不包含 ;
   • 不包含 DROP / DELETE / UPDATE / INSERT 等关键字
2. MySQL 连接释放
   • 每个导出任务会从连接池获取一个连接，并在完成后自动释放，无需手动管理。
3. Redis 用途
   • 仅用于存储任务进度，不参与核心导出逻辑。Redis 断开不影响导出，但无法查询进度。
4. 大表建议
   • 避免 SELECT *，明确指定字段
   • 添加索引到 WHERE 条件字段
   • 监控磁盘空间（GZIP 仍需临时写入）