importElasticSearch.js

(function () {
  'use strict';

  const fs = require('fs');
  const elasticsearch = require('elasticsearch');
  const esClient = new elasticsearch.Client({
    host: 'localhost:9200',
    log: 'error'
  });

  const bulkIndex = function bulkIndex(index, type, data) {
    let bulkBody = [];

    data.forEach(item => {
      bulkBody.push({
        index: {
          _index: index,
          _type: type,
          _id: item.code // we can change it as per requirement
        }
      });

      bulkBody.push(item);
    });

    esClient.bulk({ body: bulkBody })
      .then(response => {
        let errorCount = 0;
        response.items.forEach(item => {
          if (item.index && item.index.error) {
            console.log(++errorCount, item.index.error);
          }
        });
        console.log(`Successfully indexed ${data.length - errorCount} out of ${data.length} items`);
      })
      .catch(console.err);
  };

  // only for testing purposes
  // all calls should be initiated through the module
  const test = function test() {
    const csvRawData = fs.readFileSync('convertcsv.json');
    const csvRaw = JSON.parse(csvRawData);
    console.log(`${csvRaw.length} items parsed from data file`);
    bulkIndex('cities-info', 'cities', csvRaw);
  };

  test();

  module.exports = {
    bulkIndex
  };
}());