elasticsearch 7.x support

Author: cigolpl

README.md

@@ -257,6 +257,11 @@ elasticbulk.import(data, {
 ## Tests
 
 ```bash
-# It tests importing with your ES version. Don't tests on production environment
-ES_URL=http://localhost:9200 mocha tests/indexSpec.js -t 100000
+# Test ES 1.7
+docker run -it -d  -p 9200:9200 -p 9300:9300 -v $HOME/elasticsearch1.7/data:/data -v $HOME/elasticsearch1.7/logs:/logs barnybug/elasticsearch:1.7.2
+mocha --exit -t 15000 tests/elasticitemsSpec.js
+
+# Test ES 7.x
+docker run -it -d -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:7.10.1
+mocha --exit -t 15000 tests/elasticitems7xSpec.js
 ```

lib.js

@@ -3,6 +3,7 @@ const Promise = require('bluebird');
 //const isStream = require('is-stream');
 
 const elasticsearch = require('./src/elasticitems');
+const elasticsearch7x = require('./src/elasticitems7x');
 const itemsapi = require('./src/itemsapi');
 const meilisearch = require('./src/meilisearch');
 
@@ -17,6 +18,8 @@ module.exports.import = function(data, options, schema) {
     return itemsapi.import(data, options, schema);
   } else if (options.engine === 'meilisearch') {
     return meilisearch.import(data, options, schema);
+  } else if (options.engine === 'elasticsearch7x') {
+    return elasticsearch7x.import(data, options, schema);
   } else {
     return elasticsearch.import(data, options, schema);
   }

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "elasticbulk",
-  "version": "1.0.22",
+  "version": "1.0.24",
   "description": "Add data in bulk to ItemsAPI or Elasticsearch. It supports data streaming from PostgreSQL or filesystem",
   "main": "lib.js",
   "scripts": {
@@ -10,7 +10,7 @@
   "license": "MIT",
   "dependencies": {
     "bluebird": "^3.5.1",
-    "elasticsearch": "^13.3.1",
+    "elasticsearch": "^16.7.2",
     "is-stream": "^1.1.0",
     "itemsapi": "^2.1.0",
     "lodash": "^4.17.15",

package.json

@@ -0,0 +1,149 @@
+'use strict';
+
+const _ = require('lodash');
+const Promise = require('bluebird');
+const isStream = require('is-stream');
+const elasticsearch = require('elasticsearch');
+
+/**
+ * data is json array of objects or stream
+ */
+module.exports.import = function(data, options, schema) {
+
+  options = options || {}
+  options.concurrency = options.concurrency || 1
+  options.type = options.type || options.index
+
+  if (!options.host) {
+    return Promise.reject('Please define host name')
+  }
+
+  if (!options.index) {
+    return Promise.reject('Please provide index name')
+  }
+
+  if (isStream(data)) {
+    data.pause();
+  }
+
+  var elastic = new elasticsearch.Client({
+    host: options.host,
+    defer: function () {
+      return Promise.defer();
+    }
+  });
+
+  return Promise.resolve()
+  .then(function(res) {
+
+    if (schema) {
+
+      return elastic.indices.create({
+        index: options.index,
+        body: schema
+      })
+    }
+  })
+  .then(function(res) {
+
+    if (isStream(data)) {
+      return module.exports.addItemsStream(elastic, data, options)
+    } else {
+
+      return Promise.all(_.chunk(data, options.chunk_size))
+      .map(items => {
+        return module.exports.addBulkItems(elastic, items, options)
+      }, {
+        concurrency: options.concurrency
+      })
+    }
+  })
+  .catch(function(res) {
+    console.log(res);
+  })
+}
+
+/**
+ * import data by stream (file, json, psql, etc)
+ */
+module.exports.addItemsStream = function(elastic, stream, options) {
+
+
+  return new Promise(function(resolve, reject) {
+    var counter = 0;
+    var global_counter = 0;
+    var items = [];
+    var counter_limit = options.chunk_size || options.limit || 100
+    var concurrency = 1
+    var added = 1
+
+    stream.on('data', function (item) {
+
+      ++counter
+      items.push(item)
+
+      if (counter >= counter_limit) {
+        stream.pause();
+
+        return module.exports.addBulkItems(elastic, items, options)
+        .then(function(res) {
+          counter = 0;
+          items = []
+          console.log(added + ' series added!');
+          added++
+          stream.resume()
+        })
+      }
+    })
+
+    stream.on('end', function (data) {
+
+      if (!items.length) {
+        return resolve()
+      }
+
+      module.exports.addBulkItems(elastic, items, options)
+      .then(function(res) {
+        console.log('Last ' + added + ' series added!');
+        return resolve()
+      })
+    })
+
+    stream.on('close', function (data) {
+      //console.log('close');
+      return resolve()
+    })
+
+    stream.on('error', function (err) {
+      //console.log('error');
+      return reject(err)
+    })
+
+    /**
+     * it waits until creating schema is not finished
+     */
+    stream.resume();
+  })
+}
+
+module.exports.addBulkItems = function(elastic, items, options) {
+
+  var body = [];
+  for (var i = 0 ; i < items.length ; ++i) {
+    var o = { index: { _id: items[i] ? items[i]._id : undefined } };
+    body.push(o);
+    body.push(items[i]);
+  }
+
+  return elastic.bulk({
+    index: options.index,
+    //type: options.type,
+    body: body
+  })
+  .then(v => {
+
+    if (options.debug && v.errors) {
+      console.log(JSON.stringify(v, null, 2));
+    }
+  })
+}