使用npm csv-batch处理大CSV时批量执行函数未全部执行问题排查

你的核心问题在于异步流程未正确等待所有批次执行完成，同时代码存在冗余Promise封装和未绑定的异步状态，导致部分数据库插入操作被跳过或未执行。

1. 外层Promise未绑定csvBatch的完成状态 #

你创建的streamPromise没有在csvBatch处理完成时调用resolve，这会导致外层await streamPromise永远处于等待状态，在Serverless等环境中可能被提前终止，直接中断后续批次执行。

修复代码：

module.exports.handler = async (file) => {
  const streamPromise = new Promise((resolve, reject) => {      
    let stream = readStream(file);
    csvBatch(stream, {
      batch: true,
      batchSize: 2000,
      batchExecution: (batch) => insertBatchToDatabase(batch)
    }).then((results) => {
      console.log(results);
      console.log('Done!');
      resolve(results); // 关键：将csvBatch的完成状态绑定到外层Promise
    }).catch(err => {
      console.error('CSV processing failed:', err);
      reject(err); // 捕获未处理错误，避免静默失败
    });
  });
  await streamPromise;
};

2. 数据库插入函数的冗余Promise封装 #

insertBatchToDatabase中额外封装的streamPromise完全冗余，且未正确处理pgQuery的异步状态。async函数本身会返回Promise，直接等待pgQuery即可，确保csvBatch会等待当前批次插入完成后再处理下一批。

修复代码：

async function insertBatchToDatabase(rows){
  try {
    const queryString = `INSERT INTO users (${columnString}) VALUES %L`;
    const formatQuery = format(queryString, rows);
    await pgQuery(formatQuery); // 直接等待数据库操作完成
    console.log(`Inserted ${rows.length} records`); // 打印批次日志，便于排查执行情况
  } catch (err) {
    console.error(`Batch insert failed (${rows.length} records):`, err);
    throw err; // 抛出错误，让csvBatch捕获并记录到errors数组
  }
};

3. 额外排查点 #

• 确认readStream是否设置了正确编码（如readStream(file, { encoding: 'utf8' })），避免流读取异常。
• 如果CSV包含表头，需在csvBatch配置中添加headers: true，防止表头被当作数据插入，同时保证totalRecords统计准确。
• 检查数据库连接池配置：若批次执行过于频繁，可能导致连接池耗尽，后续插入失败。可调整连接池最大连接数，或确认pgClient单例连接支持批量操作。

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