Class: Elasticsearch::Model::Extensions::BatchUpdating::BatchUpdater

Inherits:
Object
  • Object
show all
Defined in:
lib/elasticsearch/model/extensions/batch_updating/batch_updater.rb

Instance Method Summary collapse

Constructor Details

#initialize(klass) ⇒ BatchUpdater

Returns a new instance of BatchUpdater.



6
7
8
 
# File 'lib/elasticsearch/model/extensions/batch_updating/batch_updater.rb', line 6

def initialize(klass)
  @klass = klass
end

Instance Method Details

#klass ⇒ Object 



10
11
12
 
# File 'lib/elasticsearch/model/extensions/batch_updating/batch_updater.rb', line 10

def klass
  @klass
end

#reconnect! ⇒ Object 



14
15
16
17
 
# File 'lib/elasticsearch/model/extensions/batch_updating/batch_updater.rb', line 14

def reconnect!
  klass.connection.reconnect!
  klass.__elasticsearch__.client = Elasticsearch::Client.new(host: klass.elasticsearch_hosts)
end

#split_ids_into(chunk_num, min: nil, max: nil) ⇒ Object 



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/elasticsearch/model/extensions/batch_updating/batch_updater.rb', line 47

def split_ids_into(chunk_num, min:nil, max:nil)
  min ||= klass.minimum(:id)
  max ||= klass.maximum(:id)
  chunk_num.times.inject([]) do |r,i|
    chunk_size = ((max-min+1)/chunk_num.to_f).ceil
    first = chunk_size * i

    last = if i == chunk_num - 1
             max
           else
             chunk_size * (i + 1) - 1
           end

    r << (first..last)
  end
end

#update_index_in_batch(records, index: nil, type: nil, client: nil) ⇒ Object

Parameters:

  • records (Array) 


20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/elasticsearch/model/extensions/batch_updating/batch_updater.rb', line 20

def update_index_in_batch(records, index: nil, type: nil, client: nil)
  client ||= klass.__elasticsearch__.client
  index ||= klass.index_name
  type ||= klass.document_type

  if records.size > 1
    response = client.bulk \
                   index:   index,
                   type:    type,
                   body:    records.map { |r| { index: { _id: r.id, data: r.as_indexed_json } } }

    one_or_more_errors_occurred = response["errors"]

    if one_or_more_errors_occurred
      if defined? ::Rails
        ::Rails.logger.warn "One or more error(s) occurred while updating the index #{records} for the type #{type}\n#{JSON.pretty_generate(response)}"
      else
        warn "One or more error(s) occurred while updating the index #{records} for the type #{type}\n#{JSON.pretty_generate(response)}"
      end
    end
  else
    records.each do |r|
      client.index index: index, type: type, id: r.id, body: r.as_indexed_json
    end
  end
end