使用JavaScript从本地文件流式批量插入PostgreSQL数据

我之前处理过类似的超大数据导入需求，逐行插入确实会因为频繁的数据库交互拖慢速度，流式+批量插入绝对是解决这个问题的最优路径。结合你提到的Node.js环境（从--max-old-space-size参数能看出来），下面给你一套落地性很强的解决方案：

核心思路 #

• 流式读取：完全避免一次性加载整个800万条数据到内存，用流式工具逐段解析JSON
• 批量攒数：每攒够10000条就触发一次数据库写入，平衡内存占用和写入效率
• 优化写入：用PostgreSQL的批量INSERT或COPY命令，这两种方式比单条插入快几个数量级

具体实现步骤（Node.js环境） #

1. 先装必要依赖 #

npm install pg readline JSONStream

• pg：PostgreSQL官方Node.js客户端，支持连接池和批量操作
• readline：Node.js内置工具，适合处理每行一个JSON对象的大文件
• JSONStream：专门用来流式解析大JSON数组（比如[{},{},...]格式的文件）

2. 情况1：JSON文件是「每行一个JSON对象」（最常用的大文件格式） #

这种格式不需要解析整个数组，直接逐行读取即可，内存占用极低：

const fs = require('fs');
const readline = require('readline');
const { Pool } = require('pg');

// 配置PostgreSQL连接池（复用连接，减少开销）
const pool = new Pool({
  user: '你的数据库用户名',
  host: '数据库地址',
  database: '目标数据库名',
  password: '数据库密码',
  port: 5432,
});

const BATCH_SIZE = 10000; // 每批插入1万条
let batchData = [];
let totalInserted = 0;

// 创建流式读取接口
const rl = readline.createInterface({
  input: fs.createReadStream('你的超大JSON文件.json'),
  crlfDelay: Infinity // 兼容不同换行符
});

// 处理每一行数据
rl.on('line', async (line) => {
  try {
    const data = JSON.parse(line);
    batchData.push(data);
    totalInserted++;

    // 攒够批次就触发插入
    if (batchData.length >= BATCH_SIZE) {
      await insertBatch(batchData);
      batchData = []; // 清空当前批次
      console.log(`已完成 ${totalInserted} 条数据插入`);
    }
  } catch (err) {
    console.error(`解析行失败，跳过该行: ${err.message}`);
  }
});

// 处理文件末尾剩余的不足一批的数据
rl.on('close', async () => {
  if (batchData.length > 0) {
    await insertBatch(batchData);
    console.log(`全部插入完成！总计 ${totalInserted} 条数据`);
  }
  await pool.end(); // 关闭连接池
});

// 批量插入核心函数（用VALUES批量INSERT）
async function insertBatch(data) {
  // 生成SQL占位符：比如($1,$2),($3,$4)... 请根据你的表字段调整
  const placeholders = data.map((_, idx) => 
    `($${idx*2+1}, $${idx*2+2})`
  ).join(',');

  // 把所有数据扁平化，对应占位符的顺序
  const values = data.flatMap(item => [item.字段1, item.字段2]);

  const query = `
    INSERT INTO 你的表名 (字段1, 字段2)
    VALUES ${placeholders}
    ON CONFLICT DO NOTHING; -- 如果不需要处理重复键，可以删掉这行
  `;

  await pool.query(query, values);
}

3. 情况2：JSON文件是「大数组格式」（比如[{},{},...,{}]） #

这种情况用JSONStream流式解析数组元素，避免加载整个大数组到内存：

const fs = require('fs');
const JSONStream = require('JSONStream');
const { Pool } = require('pg');

const pool = new Pool({ /* 连接配置同上 */ });
const BATCH_SIZE = 10000;
let batchData = [];
let totalInserted = 0;

// 流式解析JSON数组的每个元素
fs.createReadStream('你的超大JSON数组文件.json')
  .pipe(JSONStream.parse('*')) // '*'表示匹配数组里的每个元素
  .on('data', async (data) => {
    batchData.push(data);
    totalInserted++;

    if (batchData.length >= BATCH_SIZE) {
      await insertBatchWithCopy(batchData); // 用COPY命令更快
      batchData = [];
      console.log(`已完成 ${totalInserted} 条数据插入`);
    }
  })
  .on('end', async () => {
    if (batchData.length > 0) {
      await insertBatchWithCopy(batchData);
      console.log(`全部插入完成！总计 ${totalInserted} 条数据`);
    }
    await pool.end();
  })
  .on('error', (err) => {
    console.error(`解析JSON失败: ${err.message}`);
  });

// 用PostgreSQL COPY命令插入（比批量INSERT快2-3倍）
async function insertBatchWithCopy(data) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    // 把数据转换成CSV格式（COPY命令原生支持CSV）
    const csvContent = data.map(item => `${item.字段1},${item.字段2}`).join('\n');
    // 执行COPY命令
    await client.query(
      `COPY 你的表名 (字段1, 字段2) FROM STDIN WITH (FORMAT CSV)`,
      [],
      (err) => {
        if (err) throw err;
        client.write(csvContent);
        client.endCopy();
      }
    );
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release(); // 释放连接回池
  }
}

额外性能优化建议 #

• 调整批次大小：1万条是比较均衡的数值，你可以根据服务器内存微调（比如5000-20000之间）
• 临时关闭索引：如果目标表有很多索引，插入前先删除，插入完成后再重建，能大幅提升速度（注意：插入期间无法使用索引查询）
• 关闭自动提交：批量插入时手动控制事务（BEGIN/COMMIT），减少事务日志的写入次数
• 监控内存占用：可以用process.memoryUsage()在代码里监控内存，确保不会出现内存溢出

内容的提问来源于stack exchange，提问作者pir