Module: Elasticsearch::Model::Importing::ClassMethods

Included in:
Proxy::ClassMethodsProxy
Defined in:
lib/elasticsearch/model/importing.rb

Instance Method Summary collapse

Instance Method Details

#__batch_to_bulk(batch, transform) ⇒ Object 

 178 179 180
 
# File 'lib/elasticsearch/model/importing.rb', line 178 def __batch_to_bulk(batch, transform) batch.map { |model| transform.call(model) } end

#import(options = {}, &block) {|Hash| ... } ⇒ Fixnum, Array<Hash>

Import all model records into the index

The method will pick up correct strategy based on the ‘Importing` module defined in the corresponding adapter.

Examples:

Import all records into the index

 Article.import

Set the batch size to 100

 Article.import batch_size: 100

Process the response from Elasticsearch

 Article.import do |response| puts "Got " + response['items'].select { |i| i['index']['error'] }.size.to_s + " errors" end

Delete and create the index with appropriate settings and mappings

 Article.import force: true

Refresh the index after importing all batches

 Article.import refresh: true

Import the records into a different index/type than the default one

 Article.import index: 'my-new-index', type: 'my-other-type'

Pass an ActiveRecord scope to limit the imported records

 Article.import scope: 'published'

Pass an ActiveRecord query to limit the imported records

 Article.import query: -> { where(author_id: author_id) }

Transform records during the import with a lambda

 transform = lambda do |a| {index: {_id: a.id, _parent: a.author_id, data: a.__elasticsearch__.as_indexed_json}} end Article.import transform: transform

Update the batch before yielding it

 class Article # ... def self.enrich(batch) batch.each do |item| item.metadata = MyAPI.get_metadata(item.id) end batch end end Article.import preprocess: :enrich

Return an array of error elements instead of the number of errors, e.g. to try importing these records again

 Article.import return: 'errors'

Parameters:

  • options (Hash) (defaults to: {}) —

    Options passed to the underlying ‘__find_in_batches` method

  • block (Proc) —

    Optional block to evaluate for each batch

Yields:

  • (Hash) —

    Gives the Hash with the Elasticsearch response to the block

Returns:

  • (Fixnum) —

    default, number of errors encountered during importing

  • (Array<Hash>) —

    if return option is specified to be “errors”, contains only those failed items in the response items key, e.g.:

    [ { "index" => { "error" => 'FAILED', "_index" => "test", "_id" => '1', "_version" => 1, "result" => "foo", "_shards" => { "total" => 1, "successful" => 0, "failed" => 1 }, "status" => 400 } } ] 
 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176
 
# File 'lib/elasticsearch/model/importing.rb', line 136 def import(options={}, &block) errors = [] refresh = options.delete(:refresh) || false target_index = options.delete(:index) || index_name transform = options.delete(:transform) || __transform pipeline = options.delete(:pipeline) return_value = options.delete(:return) || 'count' unless transform.respond_to?(:call) raise ArgumentError, "Pass an object responding to `call` as the :transform option, #{transform.class} given" end if options.delete(:force) self.create_index! force: true, index: target_index elsif !self.index_exists? index: target_index raise ArgumentError, "#{target_index} does not exist to be imported into. Use create_index! or the :force option to create it." end __find_in_batches(options) do |batch| params = { index: target_index, body: __batch_to_bulk(batch, transform) } params[:pipeline] = pipeline if pipeline response = client.bulk params yield response if block_given? errors += response['items'].select { |k, v| k.values.first['error'] } end self.refresh_index! index: target_index if refresh case return_value when 'errors' errors else errors.size end end