使用 node.js 在 neo4j 中导入 CSV

Question

我正在尝试使用 node.js 将 csv 文件导入 neo4j。我必须将数据插入多个collection/table，所以我必须使用node.js 脚本插入数据。但我的问题是，插入csv数据时无法防止数据重复。

CSV 数据示例：

name
-------------
Afghanistan
Afghanistan
Aland
Albania
Albania
Bangladesh
Bangladesh

index.js

cp = require('child_process');
child = cp.fork(__dirname + "/background-import-csv-file.js");
child.on('message', function(msg) {
    console.log("background-insert-process said : ", msg);
});
file = path.resolve(__dirname, `./file/simplemaps.csv`);
child.send(file);

在background-import-csv-file.js，我用两种不同的方式编写代码。

基于第一个承诺 (background-import-csv-file.js)：

cp = require('child_process');
csv = require('fast-csv');
Q = require('q');
DB = require("./common/driver");
Country = require('./collection/country');
process.on("message", (file) => {
  stream = fs.createReadStream(file);
  csv
    .fromStream(stream, { headers: true })
    .on("data", function(data) {
        let countryData = { "name": data.name };
        neo = new DB();
        country = new Country(neo);
        country.insert(countryData)
            .then(resp => process.send(resp.msg) )
            .catch(err => process.send(err) )
    })
     .on("end", () => process.send("file read complete") );
});

./collection/country.js:

  Q = require('q');
  Country = function Country(neo) {
    this.country = "Country";  this.neo = neo;
  };

  Country.prototype.find = function find(filters) {
     query = `MATCH (a:Country  { name: '${filters.name}' } )  RETURN {country:properties(a)}`;
     return this.neo.run(query, filters).then(resp => resp);
  }

  Country.prototype.create = function create(data) {
    query = `CREATE (ax:Country  { name: '${data.name}' } )  RETURN ax `;
    return this.neo.run(query, {}).then(resp => resp[0].properties).catch(err => err)
   }

   Country.prototype.insert = function insert(country) {
      filter = { name: country.name };
      return Q(this.find(filter))
        .then(resp => resp.length > 0 ? Q.resolve({ msg: `country: [${country.name}] is already exist` }) : Q.resolve(this.create(country))  )
    .then(resp => resp)
    .catch(e => Q.reject(e));
   }

   module.exports = Country;

./common/driver.js

neo4j = require('neo4j-driver').v1;
function DB() {
   this.driver = neo4j.driver();   this.session = this.driver.session();
}

DB.prototype.run = function run(query, data) {
    return this.session.run(query, data)
    .then(response => response.records.map(
            record => record._fields[0] ?
            record._fields.length ? record._fields[0] : {} : {}
        ) ).catch(err => new Error(err) );
}

module.exports = DB;

当我在终端中运行 index.js 时，在数据库中，我有 2 个 Afghanistan、1 个 Aland、2 个 Albania 和 2 个 Bangladesh。但我的数据库中需要 1 个Afghanistan、1 个Aland、1 个Albania 和 1 个Bangladesh。当我分析代码时，发现在插入数据之前，我正在检查数据（Country.prototype.find = function find(filters)）是否已经存在，但它总是返回空结果。这就是为什么它插入多个数据。如果我再次运行index.js，则不会将新数据插入数据库。为了解决这个问题，我试过关注CQL：

  MERGE (c:Country  { name: '${data.name}' } )  RETURN c

它是插入唯一数据，但它会浪费很多时间。然后我写了以下代码：

事件驱动 (background-import-csv-file.js)：

process.on("message", (file) => {
  stream = fs.createReadStream(file);
  csv
    .fromStream(stream, { headers: true })
    .on("data", function(data) {
        countryData = { "name": data.name };
        neo = new DB();
        country = new Country(neo);
        country.find(countryData);
        country.on('find', resp =>  resp.length > 0 ? Q.resolve({ msg: `country: [${country.name}] is already exist` }) : Q.resolve(country.create(countryData))  );

        country.on('create', resp => console.log(resp) );
    })
    .on("end", () => process.send("file read complete") );
});

./collection/country.js:

 EventEmitter = require('events').EventEmitter;
 util = require('util');

 Country = function Country(neo) {
   this.neo = neo;  EventEmitter.call(this);
 };
 util.inherits(Country, EventEmitter);

 Country.prototype.find = function find(filters) {
    query = `MATCH (a:Country  { name: '${filters.name}' } )  RETURN {country:properties(a)}`;
    return this.neo.run(query, {}).then(resp => this.emit('find', resp));
 }

 Country.prototype.create = function create(data) {
    query = `CREATE (ax:Country  { name: '${data.name}' } )  RETURN ax `;
    return this.neo.run(query, {}).then(resp => this.emit('create', resp[0].properties)).catch(err =>  err)
 }

这一次，它显示了相同的结果。我错过了什么？任何建议都会非常有帮助。

注意：我使用 fast-csv 进行 csv 解析，使用 Q 进行承诺。

Answer 1

其实我可以想象以下解决方案：

1. 使用编程语言（如 node.js）修改 CSV 文件本身以删除具有相同名称的重复行。
2. 添加neo4j unique constrains CREATE CONSTRAINT ON (c:Country) ASSERT c.name IS UNIQUE
3. 涉及中间件，如队列，防止重复项，为此，您需要定义自己的消息结构和重复算法。

上面。

Answer 2

我的问题是，在csv 文件解析中，它是如此之快（事件驱动），以至于没有等待将数据插入数据库。所以我必须暂停文件解析然后恢复它。

我使用以下代码解决了我的问题：

基于承诺（background-import-csv-file.js）：

cp = require('child_process');
csv = require('fast-csv');
Q = require('q');
DB = require("./common/driver");
Country = require('./collection/country');

process.on("message", (file) => {
  stream = fs.createReadStream(file);
  csvstream = csv
    .fromStream(stream, { headers: true })
    .on("data", function(data) {
       csvstream.pause();  // pause the csv file parsing
       countryData = { "name": data.name };
       neo = new DB();
       country = new Country(neo);
       country.insert(countryData)
         .then(resp => {
             process.send(resp.msg);
             neo.close();
             return csvstream.resume(); // after completing db process, resume 
         })
         .catch(err => {
             process.send(err);
             return csvstream.resume();  // if failed, then resume 
          });
    })
    .on("end", () => process.send("file read complete") );
 });