Support for simply updating existing documents

.gitignore

@@ -1 +1,2 @@
 node_modules
+.idea

cmd/get-args.js

@@ -11,5 +11,6 @@ module.exports = () => yargs
     yargs
       .number('batch')
       .number('timeout')
+      .boolean('update')
   )
   .argv;

cmd/init-cmd.js

@@ -19,7 +19,7 @@ function cmdPutDocs(args) {
 
   return util.readStdin()
     .map(JSON.parse)
-    .through(_.partialRight(putDocs, args._[1], { batch: args.batch, requestTimeout: args.timeout }))
+    .through(_.partialRight(putDocs, args._[1], { batch: args.batch, requestTimeout: args.timeout, update: args.update }))
     .map(JSON.stringify);
 }
 

lib/put-docs.js

@@ -9,20 +9,26 @@ const _ = require('lodash'),
  * Transform an array of docs into an array of bulk actions.
  * @param  {Object[]} docs
  * @param  {string} index Index each doc will be put into
+ * @param  {boolean} update Indicates if the bulk operation is index or update
  * @return {Object[]}
  */
-function docsToBulkActions(docs, index) {
+function docsToBulkActions(docs, index, update = false) {
   return docs.reduce((acc, doc) => {
+    const reqOptions = { _index: index || doc._index, _type: doc._type || DEFAULT_TYPE, _id: doc._id };
+
+    if (update) {
+      acc.push(
+        { update: reqOptions, doc_as_upsert: true},
+        { doc: doc._source }
+      );
+      return acc;
+    }
+
     acc.push(
-      {
-        index: {
-          _index: index || doc._index,
-          _type: doc._type || DEFAULT_TYPE,
-          _id: doc._id
-        }
-      },
+      { index: reqOptions },
       doc._source
     );
+
     return acc;
   }, []);
 }
@@ -34,15 +40,16 @@ function docsToBulkActions(docs, index) {
  * @param  {string} elastic Elastic endpoint e.g. localhost:9200/foo
  * @param  {Object} [options]
  * @param  {number} [options.batch] Number of items to include in each bulk req
+ * @param  {boolean} [options.update] Indicates if the bulk operation is index or update
  * @return {Stream} of bulk results
  */
-function putDocs(stream, elastic, {batch = DEFAULT_BATCH_SIZE, requestTimeout} = {}) {
+function putDocs(stream, elastic, {batch = DEFAULT_BATCH_SIZE, requestTimeout, update = false} = {}) {
   const {host, index} = util.parseElastic(elastic),
     client = util.getEsClient({host, requestTimeout});
 
   return stream
     .batch(batch)
-    .map(_.partialRight(docsToBulkActions, index))
+    .map(_.partialRight(docsToBulkActions, index, update))
     .flatMap(batchActions => h(client.bulk({body: batchActions})))
     .flatMap(results => h(results.items));
 }

lib/put-docs.test.js

@@ -8,7 +8,7 @@ const fn = require('./put-docs'),
   _ = require('lodash'),
   { DEFAULT_BATCH_SIZE } = require('./constants');
 
-describe('get-docs', function () {
+describe('put-docs', function () {
   let sandbox,
     esClientStub,
     mockDocs,

package.json

@@ -1,6 +1,6 @@
 {
   "name": "elasticsearch-streamer",
-  "version": "1.2.0",
+  "version": "1.2.1",
   "description": "A programmatic and command-line utility for streaming docs from Elasticsearch indices and putting docs into indices in bulk.",
   "main": "index.js",
   "scripts": {