MapFilter importer

elasticsearch_importer.rb

# Import the data into Elasticsearch
class ElasticsearchImporter
  def call
    create_new_index
    swap_alias
    import
    delete_old_index
  end

  # == Parameters:
  # index_alias::
  #   A String declaring the index alias. Index name will be an union of alias and a timestamp.
  # mapping::
  #   A Hash with index mapping.
  #   Follow Elasticsearch official guide to get familiar with format https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping.html
  # payload::
  #   A Array with a Hashes compatible with Elasticsearch bulk API format.
  #   Follow Elasticsearch official guide to get familiar with bulk payload https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
  #
  def initialize(index_alias:, mapping:, payload:)
    @index_alias = index_alias

    @old_index_name = if ES.with { |es| es.indices.exists_alias? name: index_alias }
                        ES.with { |es| es.indices.get_alias name: index_alias }.keys.first
                      end

    @new_index_name = "#{index_alias}_#{Time.now.to_f}"
    @mapping = mapping
    @payload = payload
  end

  private

  attr_reader :index_alias, :old_index_name, :new_index_name, :payload, :mapping

  def create_new_index
    ES.with { |es| es.indices.create index: new_index_name, body: { mappings: mapping } }
  end

  def import
    ES.with { |es| es.bulk body: payload }
  end

  def swap_alias
    remove = { remove: { index: old_index_name, alias: index_alias } } if old_index_name
    create = { add: { index: new_index_name, alias: index_alias } }
    actions = [remove, create].compact

    ES.with { |es| es.indices.update_aliases body: { actions: actions } }
  end

  def delete_old_index
    return unless old_index_name

    ES.with { |es| es.indices.delete index: old_index_name }
  end
end