diff --git a/CHANGELOG.md b/CHANGELOG.md index b12dea55..9f3d184e 100644 --- a/CHANGELOG.md +++ b/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` diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb index c73a6e83..a9777090 100644 --- a/lib/elastomer/client/bulk.rb +++ b/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') # diff --git a/lib/elastomer/client/delete_by_query.rb b/lib/elastomer/client/delete_by_query.rb index d6ee0efa..52eb955a 100644 --- a/lib/elastomer/client/delete_by_query.rb +++ b/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 diff --git a/lib/elastomer/version.rb b/lib/elastomer/version.rb index 4a4028c4..b3c0e9e9 100644 --- a/lib/elastomer/version.rb +++ b/lib/elastomer/version.rb @@ -1,5 +1,5 @@ module Elastomer - VERSION = '0.6.0' + VERSION = '0.7.0' def self.version VERSION diff --git a/test/client/bulk_test.rb b/test/client/bulk_test.rb index e783b73b..f56fb109 100644 --- a/test/client/bulk_test.rb +++ b/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