Indexando dados no Elasticsearch

Neste artigo será mostrado como interagir com o Elasticsearch a partir de uma aplicação capaz de criar e indexar dados através de chamadas HTTP. De forma que o exemplo seja de simples reprodução em outras linguagens uma vez que,  nosso exemplo será feito com base em uma aplicação Ruby on Rails.

Iniciaremos com a criação de uma aplicação web Ruby on Rails que indexa dados de seu banco de dados local PostgreSQL em um índice Elasticsearch. Na sequência, iremos gerar um volume de dados que, ao final, será indexado.

Preparação do ambiente

Iniciaremos criando nossa aplicação web, que irá interagir com o Elasticsearch. Esta aplicação será uma API feita em Ruby on Rails, porém as interações com o Elasticsearch serão feitas via chamada HTTP permitindo que seja reproduzido, de forma simples, em sua linguagem favorita.

Como pré requisitos para acompanhar o passo a passo, você deve ter o Docker e docker-compose instalados e configurados em seu ambiente local. Uma vez que os pré-requisitos estejam cumpridos, podemos dar início criando nossa aplicação a partir de um container Ruby para que não haja a necessidade de que você tenha o Ruby e o Ruby on Rails instalados em seu ambiente local.

Os comandos a seguir, iniciam um container Docker a partir da imagem ruby:3.2.2. Na sequência, devemos instalar a gem Ruby on Rails que usaremos para gerar nosso projeto e, por fim, executar o comando utilizado para gerar um novo projeto rails:

# Inicia o container Ruby
docker run --rm -v $(pwd):/usr/src -w /usr/src -it ruby:3.2.2 bash
# instala o Ruby on Rails
gem install rails
# Cria um novo projeto rails
rails new reindex-blue-green --skip-test --skip-bundle --database=postgresql --api

Com nosso projeto criado, chegou a hora de configurarmos nossos containers e nossa conexão com o banco de dados para termos nosso ambiente de desenvolvimento funcional. Iremos iniciar criando um arquivo chamado Dockerfile na raiz do projeto com as seguintes configurações:

# Dockerfile
FROM ruby:3.2.2
# Instala nossas dependências
RUN apt-get update && apt-get install -qq -y --no-install-recommends build-essential \
libpq-dev git-all
# env var com o path da aplicação
ENV INSTALL_PATH /reindex-blue-green
# cria o diretório que será utilizado como WORKDIR
RUN mkdir -p $INSTALL_PATH
# define a localização inicial do container
WORKDIR $INSTALL_PATH
# copia Gemfile para o container
COPY Gemfile ./
# gems path
ENV BUNDLE_PATH /gems
# copia o diretório com o código da aplicação para nosso container
COPY . .

Agora, iremos adicionar um arquivo chamado docker-compose.yml também na raiz do projeto. Neste arquivo são definidos os seguintes serviços e volumes que iremos utilizar:

- Postgres: será o banco de dados relacional da aplicação;

- Web: aplicação web, no caso uma API Ruby on Rails;

- Elasticsearch: um container elasticsearch que usaremos para indexar nossos dados;

- Kibana: interface para interagir com o elasticsearch de forma simples.

version: "3"
services:
  postgres:
    image: "postgres:13"
    volumes:
     - postgres:/var/lib/postgresql/data
    environment:
      - POSTGRES_HOST_AUTH_METHOD: trust
    ports:
      - "5432:5432"

  web:
    depends_on:
      - "postgres"
      - "elasticsearch"
    build: .
    command: bash -c "(bundle check || bundle install) && bundle exec puma -C config/puma.rb"
    ports:
      - "3000:3000"
    volumes:
      - .:/reindex-blue-green
      - gems:/gems
   
   elasticsearch:
     image: elasticsearch:7.17.9
     depends_on:
       - postgres
     ports:
       - 9200:9200
       - 9300:9300
     volumes:
       - elasticsearch:/usr/share/elasticsearch/data
     environment:
       - ES_JAVA_OPTS: -Xms512m -Xmx512m
       - discovery.type: single-node
  
  kibana:
    image: kibana:7.17.9
    depends_on:
      - elasticsearch
    ports:
      - 5601:5601
    environment:
      - ELASTICSEARCH_HOSTS: http://elasticsearch:9200

volumes:
  postgres:
  gems:
  elasticsearch:

Após adicionar o arquivo docker-compose.yml, faremos o build e iniciaremos nossos serviços executando os seguintes comandos, a partir do diretório da aplicação:

docker build .
docker-compose up

Para concluir nosso setup, iremos configurar a conexão de nossa aplicação Ruby on Rails com nosso banco de dados PostgreSQL. Editamos o arquivo config/database.yml com o conteúdo a seguir e, na sequência, reiniciaremos nossos containers:

default: &default
  adapter: postgresql
  encoding: unicode
  pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
  host: postgres
  user: postgres
  port: 5432

development:
  <<: *default
  database: reindex_blue_green_development

test:
  <<: *default
  database: reindex_blue_green_test

Uma vez configurada e reiniciada a aplicação, acessaremos o container web e criaremos nosso banco de dados:

# Comando que acessa o container em que está sendo executado o serviço web
docker-compose exec web bash 
# Comando rails que cria o banco de dados
bundle exec rails db:create

Caso tudo tenha corrido bem ao acessar localhost:3000 veremos a seguinte tela:

Indexação

Adicionando dados para serem indexados

Nesta etapa,  criaremos um modelo de dados chamado Candidate que irá armazenar na tabela candidates os registros que iremos indexar na sequência. Nosso modelo possui os atributos name (String), email (String), experience_time(Integer) e focus (String). Para gerar nosso model e executar a migração que cria a tabela candidates, você deve executar os seguintes comandos a partir do container web:

# Cria o model Candidate
bundle exec rails g model candidate name email experience_time:integer focus
# Executa a migration pendente
bundle exec rails db:migrate

Como pretendemos gerar muitos candidatos com dados distintos, podemos utilizar a gem faker para obtermos valores "aleatórios" e que façam sentido como nomes e emails válidos. Para adicionar a gem faker adicionaremos gem 'faker' ao arquivo Gemfile no grupo development, :test e executaremos o comando bundle para que seja feita a instalação.

Com nosso modelo criado e a gem faker adicionada ao projeto, chegou o momento de popularmos nossa tabela. Faremos isso a partir de uma rake task  que criaremos em  lib/tasks/candidates.rake. Essa task faz um loop que itera cem vezes e a cada iteração um novo candidato é criado em nosso banco de dados. Note que estamos utilizando valores vindos da gem faker para que nossos candidatos tenham name e email "únicos". Também definimos que o valor de experience time será um valor entre um e dez, e o foco será um valor aleatório entre as opções back-end, front-end e full-stack. Para executar a task e criar os candidatos executaremos bundle exec rails candidates:seed.

# frozen_string_literal: true

namespace :candidates do
  desc 'Create candidates'
  task seed: :environment do
    100.times do |index|
      Candidate.create!(
        name: Faker::Name.name,
        email: Faker::Internet.email,
        experience_time: rand(1..10),
        focus: %w[backend frontend fullstack].sample
      )
      puts "#{index + 1} of 100"
    end
  end
end

Indexando dados

Agora que temos candidatos cadastrados em nossa base, chegou o momento de interagirmos com o Elasticsearch e indexá-los. Inicialmente, iremos acessar localhost:9200 e verificar se nosso container elasticsearch está funcionando. Caso esteja funcionando conforme o esperado, será exibida uma mensagem parecida com o exemplo:

Com a confirmação de que o container está funcionando corretamente, podemos dar início ao processo de indexação. Esse será feito a partir de chamadas a API que o Elasticsearch expõe para que sejam feitas operações de leitura e escrita de registros.

Apesar de o Elasticsearch possuir bibliotecas que ajudem na integração, neste artigo não será utilizada nenhuma gem para facilitar a replicação do projeto em outras linguagens.

Em um projeto real, é recomendado utilizar a biblioteca da sua linguagem favorita para realizar a integração com o Elasticsearch, uma vez que tais bibliotecas são extensivamente aplicadas em projetos reais.

Para simplificar o código que faz as chamadas HTTP, iremos utilizar a gem faraday:

# Adicionar ao arquivo Gemfile
gem "faraday"
# Comando que instala a gem
bundle

Para isolar o código que irá fazer as requests HTTP ao Elasticsearch, iremos criar uma nova classe em app/repositories/candidates_repository.rb. Esta classe seguirá o pattern Repository de forma que todas as chamadas ao Elasticsearch, relacionadas ao contexto de Candidate, serão feitas por esta classe.

Inicialmente, iremos implementar os métodos create_index que cria um novo índice com o nome recebido. Um índice pode ser interpretado como algo próximo a uma tabela em um banco de dados relacional.  Por isso, antes de adicionar registros precisamos criar um índice. Note que o Elasticsearch espera receber o parâmetro  mappings que define as propriedades e atributos do nosso índice. O outro método que iremos implementar chama-se index_candidate, que irá receber o nome de um índice e um objeto Candidate para indexá-lo.

# frozen_string_literal: true

class CandidateRepository
  # NOTE: defines the host of elasticsearch, we use elasticsearch instead of localhost due to docker network.
  ELASTICSEARCH_HOST = 'http://elasticsearch:9200/'.freeze

  # NOTE: defines the mapping with candidate attributes and its types.
  INDEX_MAPPING = {
    mappings: {
      properties: {
        name: { type: "text" },
        email: { type: "text" },
        experience_time: { type: "integer" },
        focus: { type: "text" }
      }
    }
  }.freeze

  # Create an index with received name.
  #
  # @param index_name [String] index_name
  def create_index(index_name)
    response = conn.put(index_name) do |req|
      req.body = INDEX_MAPPING.to_json
    end
    return 'Index created' if response.status == 200
    response.body
  end

  # Index candidate on received index.
  #
  # @param candidate [Candidate] candidate object.
  # @param index_name [String] index which received candidate will be indexed.
  def index_candidate(candidate, index_name)
    response = conn.put("#{index_name}/_doc/#{candidate.id}") do |req|
      req.body = {
        name: candidate.name,
        email: candidate.email,
        experience_time: candidate.experience_time,
        focus: candidate.focus
      }.to_json
    end
    return 'Candidate indexed' if response.status == 201 || response.status == 200
    response.body
  end

  private

  def conn
    Faraday.new(
      url: ELASTICSEARCH_HOST,
      headers: {'Content-Type' => 'application/json'}
    )
  end
end

Por fim, iremos adicionar uma nova rake task no arquivo lib/tasks/candidates.rake chamada candidates:mass_indexation que irá criar um índice e iterar sobre cada um de nossos candidatos, indexando-os no Elasticsearch a partir de chamadas da classe CandidateRepository.

desc 'Index candidates'
  task mass_indexation: :environment do
    index_name = 'candidates'
    puts 'Creating index ...'
    CandidateRepository.new.create_index(index_name)
    puts 'Index created'
    Candidate.all.find_each.with_index do |candidate, index|
    CandidateRepository.new.index_candidate(candidate, index_name)
    puts "#{index + 1} of #{Candidate.count}"
  end
end

Ao executar nossa nova task a partir de bundle exec rails candidates:mass_indexation podemos acessar o Kibana em http://localhost:5601 para ver os dados indexados. O Kibana é uma ferramenta utilizada para interagir com o Elasticsearch a partir de uma interface web, semelhando ao que o pgAdmin é para o PostgreSQL. Acessando http://localhost:5601/app/dev_tools#/console, você verá o editor onde podemos escrever queries para buscar dados, criar ou remover índices e mais.

As queries que iremos escrever no console do Kibana seguem a sintaxe do Elasticsearch e por isso se parecem muito com requests HTTP. A seguir, iremos executar a query GET candidates que irá nos retornar um resumo sobre o índice candidates. Note que na sessão mappings:properties temos os atributos definidos em CandidatesRepository:

{
  "candidates": {
    "aliases": {},
    "mappings": {
      "properties": {
        "email": {
          "type": "text"
        },
        "experience_time": {
          "type": "integer"
        },
        "focus": {
          "type": "text"
        },
        "name": {
          "type": "text"
        }
      }
    },
    "settings": {
      "index": {
        "routing": {
          "allocation": {
            "include": {
              "_tier_preference": "data_content"
            }
          }
        },
        "number_of_shards": "1",
        "provided_name": "candidates",
        "creation_date": "1680446848009",
        "number_of_replicas": "1",
        "uuid": "TPOOYBdMRSuRHSsGHkh0Ag",
        "version": {
          "created": "7170999"
        }
      }
    }
  }
}

Também podemos ver quantos registros temos em um índice executando a query GET candidates/_count.

{
  "count" : 100,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  }
}

E por fim, para visualizarmos os dados de candidatos indexados iremos executar a seguinte query (que é análoga a uma operação SELECT * FROM em SQL). Note que estamos utilizando o parâmetro size para que sejam retornados 100 registros uma vez que, por padrão, o Elasticsearch retorna apenas 10 registros. Com isso vemos que os dados dos nossos candidatos foram indexados corretamente!

GET candidates/_search?size=100
{ "query": { "match_all": {}} }

Conclusão

Assim chegamos ao final deste artigo, no qual vimos como criar um índice e indexar registros de forma simples no Elasticsearch, através de requests HTTP quebrando um pouco da mística envolvendo este processo.  

O código da aplicação até a conclusão deste artigo está disponível em GitHub.

Porém nem tudo são flores. Você deve ter percebido o tempo que demorou para indexar apenas cem registros e deve estar se perguntando se isso é escalável. Te trago aqui algumas observações: em projetos reais utilize a biblioteca mais ativa da sua linguagem; é possível indexar diversos registros de uma só vez, porém a indexação de uma base completa pode ser um processo demorado.

Em um segundo artigo, apresentarei uma abordagem semelhante a um deploy blue-green, que indexará nossos candidatos em um segundo índice e, ao final, efetuará a troca (switch) dos índices de forma rápida, com mínimo impacto ao usuário.

Sucesso!