csv大文件读取并且写入数据库

最近有一个需求，就是三方发来一个csv大文件（超过1个G）需要写入mongodb数据，提供接口支持自己的服务。由于文件太大，不能一次读入内存中，那么需要使用stream来进行处理，这里提供了几种方法来解决这次问题。

csv文件可以理解为一种有特定分隔符的文本文件。这样我们就可以按照文本文件对此进行处理，而不需要第三方xlsx的库了。

方案1 简单粗暴

由于mongodb是支持csv方式导入文件的，可以稍作处理，直接导入数据库。（导入前建议将相关索引都建立好）

方案2 使用readline逐行读取数据

readline作为node.js核心库，可以直接引入使用，文档地址：https://nodejs.org/dist/latest-v10.x/docs/api/readline.html

首先创建一个readline.Interface实例

const readline = require('readline');
const fs = require('fs');
const rl = readline.createInterface({
  input: fs.createReadStream('xxxx.csv')//要监听的可读流
});

监听line事件

let index = 0;//计数
let header = [];// 用于存放csv的表头
rl.on('line', async (line) => {
  rl.pause();//暂停流
  if (index === 0) {//第一行为表头
    header = _.split(line, ',');
    index++;
    // rl.prompt();
  } else {//数据行
    index++;
    let tmpData = _.split(line, ',');
    let tmpObj = _.zipObject(header, tmpData);//lodash的方法，将2个数组映射成对象
    let imei = tmpObj['muid'];
    if (imei) {
      //数据库写入操作
    } else {
      // rl.prompt();当调用时，如果 input 流已暂停，则 rl.prompt() 将恢复它。
    }
  }
});

上面的方法使用lodash的zipObject，文档如下：zipObject

刷新流，这里需要刷新流是因为上面的流被暂停了，但是这个流被暂停不是实时的。

rl.pause() 方法会暂停 input 流，允许稍后在必要时恢复它。调用 rl.pause() 不会立刻暂停 readline.Interface 实例触发的其他事件

所以我们不能再line事件中来刷新流(如果那样做的话也会将内存撑爆的，具体原因就是因为这个流暂停的不是立刻暂停的，我这里测试的是每次暂停的事件生效的时候，index已经跑到了1986了)，所以这里采用一个setInterval来刷新流，每一次刷新前都有接近2000条的数据等待着处理，我这里数据库处理2000条数据大约需要500毫秒，所以这个间隔时间我设置的500
```
setInterval(function () {
  rl.prompt();
}, 500)

rl.on('close', function () {
  console.log('close');
  // resolve(1);
});
```

这个方法并不能很好控制住数据的处理，所以也不能在线上环境使用，只有在本地作为临时处理的方案

方案3 使用readStream

使用stream的事件来处理文件，而且分隔符不仅仅限于换行分隔符(\r，\n，\r\n)。

创建流，监听事件

  let readable = fs.createReadStream(filePath);
  readable.on('open', function (fd) {
    console.log('file was opened, fd - ', fd);
  });
  readable.on('data', async function (chunk) {
    // console.log('read %d bytes: %s', chunk.length, chunk);
    await dealData(readable, chunk);
  });
  readable.on('end', function () {
    console.log('read end');
  });
  readable.on('close', function () {
    console.log('file was closed.');
  });
  readable.on('error', function (err) {
    console.log('error occured: %s', err.message);
  });

dealData处理数据

async function dealData(readable, data) {
		readable.pause();//暂停读取
		dataBuffer = Buffer.concat([dataBuffer, data]);
		let seperate = "\n";//分隔符，可以根据条件填写
		let index = dataBuffer.indexOf(seperate);
		// console.log(index)
		let lineBuffer;//行数据缓存
		let len;
		while (index > -1) {
			len = index + seperate.length;
			lineBuffer = dataBuffer.slice(0, len);//目标数据
			num++;
			//await 数据库操作lineBuffer
			//处理每一行数据
			dataBuffer = dataBuffer.slice(len, dataBuffer.length);
			index = dataBuffer.indexOf(seperate);
		}
		readable.resume();//恢复读取文件
}

目前方案3的程序运营良好，没有出现什么内存泄露的问题，能够解决目前所遇到的问题

pmlov

csv大文件读取并且写入数据库

方案1 简单粗暴

方案2 使用readline逐行读取数据

方案3 使用readStream

学习备忘录