Streaming bulk operations

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.7.0 (2015-09-18)
+- Add streaming bulk functionality via `bulk_stream_items`
+- Make Delete by Query compatible with Elasticsearch 2.0
+
 ## 0.6.0 (2015-09-11)
 - Support all URL parameters when using `Client.#scroll`
 - BREAKING: Moved some `Scroller` reader methods into `Scroller.opts`

lib/elastomer/client/bulk.rb

@@ -43,6 +43,103 @@ def bulk( body = nil, params = nil )
       end
     end
 
+    # Stream bulk actions from an Enumerator.
+    #
+    # Examples
+    #
+    #   ops = [
+    #     [:index, document1, {:_type => "foo", :_id => 1}],
+    #     [:create, document2],
+    #     [:delete, {:_type => "bar", :_id => 42}]
+    #   ]
+    #   bulk_stream_responses(ops, :index => 'default-index').each do |response|
+    #     puts response
+    #   end
+    #
+    # Returns an Enumerator of responses.
+    def bulk_stream_responses(ops, params = {})
+      bulk_obj = Bulk.new(self, params)
+
+      Enumerator.new do |yielder|
+        ops.each do |action, *args|
+          response = bulk_obj.send(action, *args)
+          yielder.yield response unless response.nil?
+        end
+
+        response = bulk_obj.call
+        yielder.yield response unless response.nil?
+      end
+    end
+
+    # Internal: Determine whether or not a response item has an HTTP status code
+    # in the range 200 to 299.
+    #
+    # item - The bulk response item
+    #
+    # Returns a boolean
+    def is_ok?(item)
+      item.values.first["status"].between?(200, 299)
+    end
+
+    # Stream bulk actions from an Enumerator and passes the response items to
+    # the given block.
+    #
+    # Examples
+    #
+    #   ops = [
+    #     [:index, document1, {:_type => "foo", :_id => 1}],
+    #     [:create, document2],
+    #     [:delete, {:_type => "bar", :_id => 42}]
+    #   ]
+    #   bulk_stream_items(ops, :index => 'default-index') do |item|
+    #     puts item
+    #   end
+    #
+    #   # return value:
+    #   # {
+    #   #   "took" => 256,
+    #   #   "errors" => false,
+    #   #   "success" => 3,
+    #   #   "failure" => 0
+    #   # }
+    #
+    #   # sample response item:
+    #   # {
+    #   #   "delete": {
+    #   #     "_index": "foo",
+    #   #     "_type": "bar",
+    #   #     "_id": "42",
+    #   #     "_version": 3,
+    #   #     "status": 200,
+    #   #     "found": true
+    #   #   }
+    #   # }
+    #
+    # Returns a Hash of stats about items from the responses.
+    def bulk_stream_items(ops, params = {})
+      stats = {
+        "took" => 0,
+        "errors" => false,
+        "success" => 0,
+        "failure" => 0
+      }
+
+      bulk_stream_responses(ops, params).each do |response|
+        stats["took"] += response["took"]
+        stats["errors"] |= response["errors"]
+
+        response["items"].each do |item|
+          if is_ok?(item)
+            stats["success"] += 1
+          else
+            stats["failure"] += 1
+          end
+          yield item
+        end
+      end
+
+      stats
+    end
 
     # The Bulk class provides some abstractions and helper methods for working
     # with the ElasticSearch bulk API command. Instances of the Bulk class
@@ -156,7 +253,7 @@ def create( document, params )
       end
 
       # Add an update action to the list of bulk actions to be performed when
-      # the bulk API call is made. Parameters can be provided in the parameters 
+      # the bulk API call is made. Parameters can be provided in the parameters
       # hash (underscore prefix optional) or in the document hash (underscore
       # prefix required).
       #
@@ -178,7 +275,7 @@ def update( document, params )
       # the bulk API call is made.
       #
       # params - Parameters for the delete action (as a Hash)
-      #      
+      #
       # Examples
       #   delete(:_id => 1, :_type => 'foo')
       #

lib/elastomer/client/delete_by_query.rb

@@ -90,35 +90,28 @@ def categorize(item)
         }
       end
 
-      # Internal: Combine the items in a response with the existing statistics
+      # Internal: Combine a response item with the existing statistics
       #
-      # response - A bulk response
-      def accumulate(response)
-        unless response.nil?
-          @response_stats['took'] += response['took']
-
-          response['items'].each do |item|
-            item = item['delete']
-            (@response_stats['_indices'][item['_index']] ||= {}).merge!(categorize(item)) { |_, n, m| n + m }
-            @response_stats['_indices']['_all'].merge!(categorize(item)) { |_, n, m| n + m }
-            @response_stats['failures'] << item unless is_ok? item['status']
-          end
-        end
+      # item - A bulk response item
+      def accumulate(item)
+        item = item["delete"]
+        (@response_stats['_indices'][item['_index']] ||= {}).merge!(categorize(item)) { |_, n, m| n + m }
+        @response_stats['_indices']['_all'].merge!(categorize(item)) { |_, n, m| n + m }
+        @response_stats['failures'] << item unless is_ok? item['status']
       end
 
       # Perform the Delete by Query action
       #
       # Returns a Hash of statistics about the bulk operation
       def execute
-        # accumulate is called both inside and outside the bulk block in order
-        # to capture bulk responses returned from calls to `delete` and the call
-        # to `bulk`
-        accumulate(@client.bulk(@params) do |bulk|
+        ops = Enumerator.new do |yielder|
           @client.scan(@query, @params).each_document do |hit|
-            accumulate(bulk.delete(_id: hit["_id"], _type: hit["_type"], _index: hit["_index"]))
+            yielder.yield([:delete, { _id: hit["_id"], _type: hit["_type"], _index: hit["_index"] }])
           end
-        end)
+        end
 
+        stats = @client.bulk_stream_items(ops, @params) { |item| accumulate(item) }
+        @response_stats['took'] = stats['took']
         @response_stats
       end
 

lib/elastomer/version.rb

@@ -1,5 +1,5 @@
 module Elastomer
-  VERSION = '0.6.0'
+  VERSION = '0.7.0'
 
   def self.version
     VERSION

test/client/bulk_test.rb

@@ -293,4 +293,31 @@
     assert_equal 'tweet', items[0]['index']['_type']
     assert_equal 1, items[0]['index']['_version']
   end
+
+  it 'streams bulk responses' do
+    ops = [
+      [:index, { :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name }]
+    ]
+    responses = $client.bulk_stream_responses(ops, { :action_count => 2 }).to_a
+    assert_equal(2, responses.length)
+    assert_bulk_index(responses[0]["items"][0])
+    assert_bulk_index(responses[0]["items"][1])
+    assert_bulk_index(responses[1]["items"][0])
+  end
+
+  it 'streams bulk items' do
+    ops = [
+      [:index, { :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name }]
+    ]
+    items = []
+    stats = $client.bulk_stream_items(ops, { :action_count => 2 }) { |item| items << item }
+    assert_equal(3, items.length)
+    assert_bulk_index(items[0])
+    assert_bulk_index(items[1])
+    assert_bulk_index(items[2])
+  end
 end