Indexando datos en Elasticsearch

Este artículo mostrará cómo interactuar con Elasticsearch desde una aplicación capaz de crear e indexar datos a través de llamadas HTTP. Para que el ejemplo se pueda reproducir fácilmente en otros lenguajes, nuestro ejemplo estará basado en una aplicación Ruby on Rails.

Comenzaremos creando una aplicación web Ruby on Rails que indexe datos de tu base de datos PostgreSQL local en un índice de Elasticsearch. A continuación, generaremos un volumen de datos que, al final, quedarán indexados.

Preparación del ambiente

Comenzaremos creando nuestra aplicación web que interactuará con Elasticsearch. Esta aplicación será una API hecha en Ruby on Rails, sin embargo las interacciones con Elasticsearch se realizarán mediante una llamada HTTP, permitiendo reproducirla, de manera sencilla, en tu lenguaje favorito.

Como requisitos previos para seguir el paso a paso, debes tener Docker y docker-compose instalados y configurados en tu entorno local. Una vez que se cumplan los requisitos previos, podemos crear nuestra aplicación desde un contenedor Ruby para que no sea necesario tener Ruby y Ruby on Rails instalados.

Los siguientes comandos inician un contenedor Docker desde la imagen ruby:3.2.2. A continuación, debemos instalar la gema Ruby on Rails que usaremos para generar nuestro proyecto y, finalmente, ejecutar el comando utilizado para generar un nuevo proyecto Rails:

# Inicia el container Ruby
docker run --rm -v $(pwd):/usr/src -w /usr/src -it ruby:3.2.2 bash
# instala Ruby on Rails
gem install rails
# Crea un nuevo proyecto rails
rails new reindex-blue-green --skip-test --skip-bundle --database=postgresql --api

Con nuestro proyecto creado, llega el momento de configurar nuestros contenedores y nuestra conexión a la base de datos para tener nuestro entorno de desarrollo funcional. Comenzaremos creando un archivo llamado Dockerfile en la raíz del proyecto con la siguiente configuración:

# Dockerfile
FROM ruby:3.2.2
# Instala nuestras dependencias
RUN apt-get update && apt-get install -qq -y --no-install-recommends build-essential \
libpq-dev git-all
# env var con el path de la aplicación
ENV INSTALL_PATH /reindex-blue-green
# crea el directorio que será utilizado como WORKDIR
RUN mkdir -p $INSTALL_PATH
# define la ubicación inicial del container
WORKDIR $INSTALL_PATH
# copia Gemfile para el container
COPY Gemfile ./
# gems path
ENV BUNDLE_PATH /gems
# copia el directorio con el código de la aplicación para nuestro container
COPY . .

Ahora agregaremos un archivo llamado docker-compose.yml también a la raíz del proyecto. Este archivo define los siguientes servicios y volúmenes que usaremos:

- Postgres: será la base de datos relacional de la aplicación;

- Web: aplicación web, en este caso una API de Ruby on Rails;

- Elasticsearch: um contenedor elasticsearch que usaremos para indexar nuestros datos;

- Kibana: Interfaz para interactuar con elasticsearch de forma sencilla.

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:

Después de agregar el archivo docker-compose.yml, compilaremos e iniciaremos nuestros servicios ejecutando los siguientes comandos, desde el directorio de la aplicación:

docker build .
docker-compose up

Para completar nuestra configuración, configuraremos la conexión entre nuestra aplicación Ruby on Rails y nuestra base de datos PostgreSQL. Editamos el archivo config/database.yml con el siguiente contenido y luego reiniciamos nuestros contenedores:

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

Una vez configurada y reiniciada la aplicación accederemos al contenedor web y crearemos nuestra base de datos:

# Comando que accede al container donde se ejecuta el servicio web
docker-compose exec web bash 
# Comando rails que crea la base de datos
bundle exec rails db:create

Si todo ha ido bien al acceder a localhost:3000 veremos la siguiente pantalla:

Indexación

Agregando datos para ser indexados

En este paso, crearemos un modelo de datos llamado Candidate que almacenará en la tabla candidates los registros que indexaremos en la secuencia. Nuestro modelo posee los atributos name (String), email (String), experience_time(Integer) y focus (String). Para generar nuestro modelo y realizar la migración que crea la tabla candidates, debes ejecutar los siguientes comandos a partir del container web:

# Crea el model Candidate
bundle exec rails g model candidate name email experience_time:integer focus
# Ejecuta la migration pendiente
bundle exec rails db:migrate

Como pretendemos generar muchos candidatos con datos diferentes, podemos utilizar la gema faker para obtener valores "aleatorios" y que tengan sentido, como nombres e emails válidos. Para agregar la gema faker, añadiremos gem faker al archivo Gemfile en el grupo development, :test y ejecutaremos el comando bundle para que se realice la instalación.

Creado nuestro modelo y añadida la gema faker al proyecto, llegó el momento de llenar nuestra tabla. Eso lo haremos a partir de una rake task que crearemos en lib/tasks/candidates.rake. Esa task hace un loop que itera cien veces y con cada iteración se crea un nuevo candidato en nuestra base de datos. Ten en cuenta que estamos usando valores de la gema faker para que nuestros candidatos tengan un nombre y correo electrónico "únicos". También definimos que el valor de experience time será de entre uno y diez, y el foco será un valor aleatorio entre las opciones back-end, front-end y full-stack. Para ejecutar la task y crear los candidatos, ejecutaremos 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 datos

Ahora que tenemos candidatos registrados en nuestra base de datos, es momento de interactuar con Elasticsearch e indexarlos. Inicialmente accederemos localhost:9200 y verificaremos si nuestro container elasticsearch está funcionando. Si se desempeña conforme a lo esperado, será mostrará un mensaje parecido al ejemplo:

Con la confirmación de que el contenedor funciona correctamente, podemos comenzar el proceso de indexación. Esto se hará mediante llamadas API que Elasticsearch expone para realizar operaciones de lectura y escritura de registros.

Aunque Elasticsearch tiene bibliotecas que ayudan con la integración, en este artículo no se utilizarán gemas para facilitar la replicación del proyecto en otros idiomas.

En un proyecto real, se recomienda utilizar su biblioteca de idioma favorita para integrar con Elasticsearch, ya que dichas bibliotecas se aplican ampliamente en proyectos reales.

Para simplificar el código que realiza llamadas HTTP, usaremos la gema faraday:

# Añadir al archivo Gemfile
gem "faraday"
# Comando que instala la gem
bundle

Para aislar el código que realizará solicitudes HTTP a Elasticsearch, crearemos una nueva clase en app/repositories/candidates_repository.rb. Esta clase seguirá el patrón Repository para que todas las llamadas a Elasticsearch, relacionadas con el contexto de Candidate sean realizadas por esta clase.

Inicialmente, implementaremos los métodos create_index que crean un nuevo índice con el nombre recibido. Un índice puede interpretarse como algo próximo a una tabla en una base de datos relacional.  Por eso, antes de agregar registros, precisamos crear un índice. Nota que Elasticsearch espera reciber el parámetro mappings que define las propiedades y atributos de nuestro índice. El otro método que implementaremos se llama index_candidate, que recibirá el nombre de un índice y un objeto Candidate para indexarlo.

# 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

Finalmente, agregaremos una nueva rake task al archivo lib/tasks/candidates.rake llamada candidates:mass_indexation, que creará un índice e iterar sobre cada uno de nuestros candidatos, indexándolos en Elasticsearch a partir de llamadas de la clase 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

Al ejecutar nuestra nueva task a partir de bundle exec rails candidates:mass_indexation, podemos accesar a Kibana en http://localhost:5601 para ver los datos indexados. Kibana es una herramienta utilizada para interactuar con Elasticsearch a partir de una interfaz web, similar a la que pgAdmin tiene con PostgreSQL. Ingresando a http://localhost:5601/app/dev_tools#/console, verás el editor donde podemos escribir queries para buscar datos, crear o remover índices y más.

Las queries que escribiremos en la consola de Kibana siguen la sintaxis de Elasticsearch, razón por la cual se parecen mucho a los requests HTTP. A continuación, ejecutaremos la query GET candidates que devolverá un resumen sobre el índice candidates. Nota que en la sesión mappings:properties tenemos los atributos definidos en 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"
        }
      }
    }
  }
}

También podemos ver cuántos registros tenemos en un índice ejecutando el query GET candidates/_count.

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

Y finalmente, para visualizar los datos candidatos indexados ejecutaremos la siguiente consulta (que es análoga a una operación SELECT * FROM en SQL). Recuerda que utilizamos el parámetro size para que se devuelvan 100 registros una vez que, por patrón, Elasticsearch retorne apenas 10 registros. ¡Con esto vemos que los datos de nuestros candidatos fueron indexados correctamente!

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

Conclusión

Así llegamos al final de este artículo, en el que vimos cómo crear un índice y registros de índice de forma sencilla en Elasticsearch, a través de solicitudes HTTP, rompiendo parte de la mística que rodea a este proceso.

El código de aplicación hasta la finalización de este artículo está disponible en GitHub.

Sin embargo, no todo es perfecto. Es posible que hayas notado cuánto tiempo llevó indexar solo cien registros y te preguntarás si esto es escalable. Aquí hay algunas observaciones: en proyectos reales, utiliza la biblioteca más activa en tu lenguaje. Es posible indexar varios registros a la vez, pero indexar una base de datos completa puede ser un proceso que requiere mucho tiempo.

En un segundo artículo, presentaré un enfoque similar a una deploy blue-green, que indexará nuestros candidatos en un segundo índice y, al final, los intercambiará (switch) de forma rápida, com mínimo impacto en el usuario.

Saludos.