Conventional database queries often aren't totally reliable in getting accurate search results fast. But fortunately, Elasticsearch comes to the rescue.

In this article, I'll walk you through how to use Elasticsearch to enhance database searches and analytics while still maintaining efficiency.

Here are the prerequisites for this tutorial:

• A Node.js environment

• Basic backend knowledge

With that, let's get started. But first of all, what is Elasticsearch?

Table of Content

• What is Elasticsearch?

• Elasticsearch Key Terms

• How to Set Up Elasticsearch

• How to Set Up the Demo Project

• How to Set Up Elasticsearch in Your Project

• How to Work with Indexes in Elasticsearch

• Search Implementation

• Full Code

• Wrapping Up

• Conclusion

What is Elasticsearch?

Elasticsearch is a search engine built by Apache that can index words and phrases, providing advanced text and vector search capabilities. It also has other useful features such as search analytics and an auto-complete feature.

Note that Elasticsearch isn't a database, even though it does provide indexing features (which popular databases also do).

Other popular alternatives to this tool used in production environments include Algolia, OpenSearch and MeiliSearch.

Elasticsearch Key Terms

in this section, we'll go over some important terminology used in Elasticsearch. To ease your understanding, I'll make references to common database terminologies.

• Index: This serves as a storage location for the data you're going to explore. It's like the database for Elasticsearch. It also shares other properties that DBs possess like uniqueness.

• Document: This is the smallest unit of information stored within the index. It's structurally identical to the MongoDB-based document and is also similar to rows in SQL-based databases.

• Mapping: Mapping refers to sets of rules or instructions that define how documents and fields are stored in the Elasticsearch index.

• Score: This is generated by Elasticsearch to show the degree of relevance of the search query to the stored index.

• Analyzer: When data is sent to the Elasticsearch engine for indexing, it initially passes through an analyzer which processes the text before indexing. This is achieved via Filters and Tokenizers.

• Tokenizers: This tool converts the gross unstructured data sent to the Elasticsearch engine into structured data tokens for further processing and storage.

• Aggregator: This search tool performs detailed analysis on the tokens stored in the index to generate actionable data insights. It's an advantage of the Elasticsearch engine. Mongo DB’s aggregator offer similar functions.

• Filter: A set of instructions which modifies tokens generated during the process of analysis. This could entail removal of fillers, capitalization rules, and so on.

• Bulk index: This refers to indexing more than one document at once. You typically do this when indexing a database with pre-existing content.

How to Set Up Elasticsearch

For the purpose of this tutorial, we'll use Elasticsearch's installable software on our local machine. Online hosted versions of Elasticsearch also exist which work hitch-free as well.

Here is a link detailing how to setup Elasticsearch on Windows. For non-Windows users, you can also install Elasticsearch on Linux/Mac OS or use Docker.

Note that for Windows users, make sure you run Elasticsearch as an Administrator to avoid installation errors.

After successful installation, you can test if it's functioning by navigating to localhost:9200 which serves as the default local endpoint for Elasticsearch. There you'll see a success message on the screen similar to the image below:

With that , we'll move on to setting up our project and integrating ElasticSearch into our demo project.

How to Set Up the Demo Project

For the sake of this tutorial, we will be utilizing a ready-built forum-based backend application built in Node Express JS.  Here is the link to the project.

to get the project up and running, clone this package and run

npm start

MySQL will serve as the default database for this tutorial.  Let's now proceed to the next section.

How to Set Up Elasticsearch in Your Project

The existing demo project is a backend implementation of a forum site which allows users to post text content and facilitate discussions through category-based threads.

Elasticsearch is great for ensuring that users can sift through these posts and threads to accurately locate key content using distinct keywords. This is more effective than using traditional database search queries which can be cumbersome.

To set up Elasticsearch, start by installing the Elasticsearch npm package. To do this, run the command below in your project directory:

npm install @elastic/elasticsearch

After successful installation, create a config.js file where you'll setup your driver to connect to your Elasticsearch application.

const { Client } = require('@elastic/elasticsearch');

const esClient = new Client({
  node: 'http://localhost:9200',
  auth: {
    username: process.env.ELASTICSEARCH_USERNAME,
    password: process.env.ELASTICSEARCH_PASSWORD
  },
  maxRetries: 5,
  requestTimeout: 60000,
  tls: {
    rejectUnauthorized: process.env.NODE_ENV !== 'development'
  }
});

module.exports = esClient;

To access and use Elasticsearch's capabilities within your backend application, you'll need to setup and configure your Elasticsearch driver. The details are specified in the config file code above.

As mentioned earlier, Elasticsearch runs on the localhost:9200 port. So your Elasticsearch node will be directed to the localhost port. Online hosted Elasticsearch nodes will also work in similar scenarios.

Next in the config file, you'll provide the authentication credentials required to access Elasticsearch. The requested username and password will be supplied within the Auth object. If you're running Elasticsearch locally, authentication may not be required unless security is enabled.

In this scenario, MaxRetries refers to the number of maximum unsuccessful attempts to access Elasticsearch. In this case, we've pegged it at 5 attempts. requestTimeout is the time in milliseconds after which the request will automatically terminate if it's not processed.

Once you've completion the Config file, you'll import this config and initialize the Elasticsearch client when your backend starts.

How to Work with Indexes in Elasticsearch

Before we start harnessing the full power of Elasticsearch, we need to customize its search capabilities within the backend of the project. This involves setting up an index within the Elasticsearch Engine that indexes all posts made to the backend application.

const esClient = require('./config');

const setupIndex = async () => {
  try {
    const indexExists = await esClient.indices.exists({
      index: INDEX_NAME
    });

    if (indexExists) {
      console.log(`Index "${INDEX_NAME}" already exists`);
      return;
    }

    await esClient.indices.create({
      index: INDEX_NAME,
      ...indexMapping
    });

    console.log(`Index "${INDEX_NAME}" created`);
  } catch (err) {
    console.error(err);
    throw err;
  }
};

The code above highlights creating a new index. First, you need to invoke the setupIndex() function. Within this function, you're providing the preferred name for your index.  Elasticsearch then checks if the name already exists within its indexes.

The function terminates if the index name already exists (to prevent duplication). But if it doesn't exist, it proceeds to create an index with that unique name alongside the index Mapping rules (which we'll discuss further shortly).

After creating the index, you'll see a success message in your application console.

How to Delete an Index

After a while, an index may no longer serve its purpose and you may need to remove it from Elasticsearch.

You can do this by executing the esClient.indices.delete() command as shown below:

const deleteIndex = async () => {
  try {
    await esClient.indices.delete({ index: INDEX_NAME });
    console.log(`${INDEX_NAME} deleted`);
  } catch (err) {
    console.error("Error deleting index:", err);
  }
};

How to Delete a Post within an Index

Sometimes, posts get deleted and modified. Also, users may get banned, after which you'd want to remove their content from the stored database .

In these cases, you'll want to ensure true deletion – that is, both from the database and from Elasticsearch indexed storage.

To do this, you'll call the esClient.delete() function, passing the Elasticsearch Client ID and the post's unique ID that you want to delete as callback arguments to your esClient.delete function.

const deletePost = async (postId) => {
  try {
    await esClient.delete({
      index: INDEX_NAME,
      id: postId.toString(),
    });

    console.log("Post successfully deleted");
    return { success: true, postId };
  } catch (err) {
    console.error(err);
    throw err;
  }
};

How to Index a Post

After setting up the Elasticsearch Index, you'll want to automatically index posts made to the database into the Elasticsearch index.

To do this, you'll need to make sure that the post is compatible with your index schema via the transformPostTOESRepo function. This function extracts and formats the post data so it matches the Elasticsearch document structure.

const transformPostToESDoc = (post) => {
  return {
  id: post.id,
  title: post.title,
  content: post.body,
  author: post.author,
  category: post.category,
  tags: post.tags,
  views: post.views || 0,
  published_at: post.created_at
};

const indexPost = async (postId) => {
  try {
    const postRepo = await getPostRepo();
    const post = await postRepo.findOne({ where: { id: postId } });

    if (!post) {
      throw new Error("Post not available");
    }

    const esDocument = transformPostToESDoc(post);

    await esClient.index({
      index: INDEX_NAME,
      id: post.id.toString(),
      document: esDocument
    });

    console.log("Post successfully indexed");
    return { success: true, postId };
  } catch (err) {
    console.error(err);
    throw err;
  }
};

The post to be indexed must have a unique ID. For ease of use, we used the unique post ID constraint that comes by default in regular databases. Optionally, you can also use UUID libraries to generate unique post IDs.

The Post information is then attached to the esClient.index() function as the document to be indexed. We also put appropriate error handling measures in place to prevent the app from crashing if the process is unsuccessful.

How to Define Elastic Search Mapping Rules

Elasticsearch mappings define how your data is stored and indexed. They specify the data type of each field and how text is analyzed for search.

In the example below, we'll define an index configuration that includes custom analyzers for autocomplete and mappings for each post field (like title, content, and author).

const indexMapping = {
  settings: {
    analysis: {
      analyzer: {
        autocomplete: {
          type: 'custom',
          tokenizer: 'standard',
          filter: ['lowercase', 'autocomplete_filter']
        },
        autocomplete_search: {
          type: 'custom',
          tokenizer: 'standard',
          filter: ['lowercase']
        }
      },
      filter: {
        autocomplete_filter: {
          type: 'edge_ngram',
          min_gram: 2,
          max_gram: 10
        }
      }
    }
  },
  mappings: {
    properties: {
      id: { type: 'integer' },
      title: {
        type: 'text',
        analyzer: 'autocomplete',
        search_analyzer: 'autocomplete_search',
        fields: {
          keyword: { type: 'keyword' },
          standard: { type: 'text' }
        }
      },
      content: {
        type: 'text',
        analyzer: 'standard'
      },
      category: {
        type: 'keyword'
      },
      tags: { type: 'keyword' },
      author: {
        type: 'text',
        fields: {
          keyword: { type: 'keyword' }
        }
      },
      views: { type: 'integer' },
      published_at: { type: 'date' }
    }
  }
};

The indexMapping object defines how Elasticsearch should store and process your data. It consists of two main parts: settings and mappings.

The mappings section defines the structure of your documents. Each field (like title, content, or author) has a type such as text, keyword, integer, or date. This tells Elasticsearch how to store and search that field.

For text fields, we can also define analyzers. Analyzers control how text is broken into smaller pieces (tokens) during indexing and search.

In the settings section, we defined a custom analyzer for autocomplete. This uses an edge_ngram filter to generate partial word matches, so users can find results as they type. We also defined a separate search_analyzer to ensure that search queries are processed correctly.

Together, these settings allow you to support features like autocomplete while keeping search results accurate and efficient.

Search Implementation

In order to implement your search functionality, you'll need to build out the API. This involves building the business logic service and the API route. You'll also use GET requests and attach your search term as a query. The result it generates will be received as a JSON document.

Then you'll implement the search post service function. In this scenario, you'll be using the search engine capabilities to search for phrases within the index. In line with best practices, you'll use a pagination technique to minimize receiving unwanted information.

The search query will consist of the index name, pagination parameters (from and size) to control which results are returned, and the expected maximum size of the result.  You'll also attach a query object specifying the modality of the search that the Elasticsearch engine should use.

const searchElastic = async (query, page = 1, size = 10) => {
  const searchQuery = {
    index: INDEX_NAME,
    from: (page - 1) * size,
    size,
    query: {
      bool: {
        must: [
          {
            multi_match: {
              query,
              fields: ["title^3", "content"],
              type: "best_fields",
              fuzziness: "AUTO"
            }
          }
        ]
      }
    }
  };

  const result = await esClient.search(searchQuery);
  return result.hits.hits;
};

In the code above, the function is named searchElastic. The function contains three variables which must be passed in order to execute it: size, page and query.

The size variable specifies the maximum number of documents per search query to be returned. The default count could be any integer.

The query uses a multi_match clause to search across multiple fields, such as title and content. The title^3 syntax boosts matches in the title, making them more relevant than matches in other fields.

We also included a must clause which defines conditions that documents must match to be included in the results.

The search results are usually ranked based on their degree of relevance to the search query.

Full Code

With this, you've completed this tutorial and have configured Elasticsearch to index posts made to your database. Here's the full code:

1. Elasticsearch Client (config.js):

const { Client } = require('@elastic/elasticsearch');

const esClient = new Client({
  node: 'http://localhost:9200',
  auth: {
    username: process.env.ELASTICSEARCH_USERNAME,
    password: process.env.ELASTICSEARCH_PASSWORD
  },
  maxRetries: 5,
  requestTimeout: 60000,
  tls: {
    rejectUnauthorized: process.env.NODE_ENV !== 'development'
  }
});

module.exports = esClient;

1. Index mapping:

const indexMapping = {
  settings: {
    analysis: {
      analyzer: {
        autocomplete: {
          type: 'custom',
          tokenizer: 'standard',
          filter: ['lowercase', 'autocomplete_filter']
        },
        autocomplete_search: {
          type: 'custom',
          tokenizer: 'standard',
          filter: ['lowercase']
        }
      },
      filter: {
        autocomplete_filter: {
          type: 'edge_ngram',
          min_gram: 2,
          max_gram: 10
        }
      }
    }
  },
  mappings: {
    properties: {
      id: { type: 'integer' },
      title: {
        type: 'text',
        analyzer: 'autocomplete',
        search_analyzer: 'autocomplete_search',
        fields: {
          keyword: { type: 'keyword' },
          standard: { type: 'text' }
        }
      },
      content: {
        type: 'text',
        analyzer: 'standard'
      },
      category: {
        type: 'keyword'
      },
      tags: { type: 'keyword' },
      author: {
        type: 'text',
        fields: {
          keyword: { type: 'keyword' }
        }
      },
      views: { type: 'integer' },
      published_at: { type: 'date' }
    }
  }
};

1. Create index:

const setupIndex = async () => {
  try {
    const indexExists = await esClient.indices.exists({
      index: INDEX_NAME
    });

    if (indexExists) {
      console.log(`Index "${INDEX_NAME}" already exists`);
      return;
    }

    await esClient.indices.create({
      index: INDEX_NAME,
      ...indexMapping
    });

    console.log(`Index "${INDEX_NAME}" created`);
  } catch (err) {
    console.error(err);
    throw err;
  }
};

1. Delete index:

const deleteIndex = async () => {
  try {
    await esClient.indices.delete({ index: INDEX_NAME });
    console.log(`${INDEX_NAME} deleted`);
  } catch (err) {
    console.error("Error deleting index:", err);
  }
};

1. Delete document (post):

const deletePost = async (postId) => {
  try {
    await esClient.delete({
      index: INDEX_NAME,
      id: postId.toString()
    });

    console.log("Post successfully deleted");
    return { success: true, postId };
  } catch (err) {
    console.error(err);
    throw err;
  }
};

1. Transform and index post:

const transformPostToESDoc = (post) => {
  return {
  id: post.id,
  title: post.title,
  content: post.body,
  author: post.author,
  category: post.category,
  tags: post.tags,
  views: post.views || 0,
  published_at: post.created_at
};

const indexPost = async (postId) => {
  try {
    const postRepo = await getPostRepo();
    const post = await postRepo.findOne({ where: { id: postId } });

    if (!post) {
      throw new Error("Post not available");
    }

    const esDocument = transformPostToESDoc(post);

    await esClient.index({
      index: INDEX_NAME,
      id: post.id.toString(),
      document: esDocument
    });

    console.log("Post successfully indexed");
    return { success: true, postId };
  } catch (err) {
    console.error(err);
    throw err;
  }
};

1. Search function:

const searchElastic = async (query, page = 1, size = 10) => {
  const searchQuery = {
    index: INDEX_NAME,
    from: (page - 1) * size,
    size,
    query: {
      bool: {
        must: [
          {
            multi_match: {
              query,
              fields: ["title^3", "content"],
              type: "best_fields",
              fuzziness: "AUTO"
            }
          }
        ]
      }
    }
  };

  const result = await esClient.search(searchQuery);
  return result.hits.hits;
};

Wrapping Up

Now you know how to use Elasticsearch to improve user search in your web applications. Elasticsearch is agnostic which allows you to use it across programming languages and frameworks. Its large community base also provides helpful user guides to make onboarding easier.

To further harness Elasticsearch's power, you can explore other tools within the ELK stack (Elasticsearch, Log Stash, and Kibana ) that'll help you generate high quality data visualizations for your data, especially for enterprise applications.

Conclusion

A fast and reliable search engine isn’t negotiable in your web applications these days. Elasticsearch is your go-to for getting this done.

If you would like to read other articles that will enhance your tech journey, feel free to check out my website here . Stay active!

We just published a comprehensive course on the freeCodeCamp.org YouTube channel designed for beginners who want to understand Elasticsearch from the ground up. Created by 3CodeCampers, this course offers a perfect mix of theory and hands-on practice. You'll start by learning the fundamentals of Elasticsearch, such as index management, document storage, text analysis, and search functionality. Then, you'll move on to advanced topics like semantic search, embeddings, and pipelines. The second part of the course focuses on applying your new skills by building a real-world project: a search engine for the Astronomy Picture of the Day (APOD) dataset.

What You'll Learn in This Course

The course is split into two parts to provide a complete learning experience:

Part 1: Elasticsearch Fundamentals

In the first part, you'll dive deep into the essential concepts of Elasticsearch, including:

• Index Management: Learn how to create and manage indexes to organize your data effectively.

• Document Storage: Understand how to store and retrieve documents using Elasticsearch APIs.

• Text Analysis and Tokenization: Discover how Elasticsearch breaks down text for powerful search capabilities.

• Search API: Learn how to perform simple and advanced searches, including filtering and aggregation.

• Semantic Search and Embeddings: Explore how to incorporate semantic search and dense vector embeddings for more relevant results.

• Pipelines and Ingest Processors: Automate data processing before storing it in Elasticsearch.

• Advanced Features: Delve into deep pagination, SQL search API, and more.

This foundational knowledge is applicable to any programming language, but the course uses Python for demonstrations, making it easy to follow along.

Part 2: Real-World Project – Build a Search Engine for APOD

In the second part, you’ll apply everything you've learned by building a practical project. You’ll create a search engine for NASA's Astronomy Picture of the Day (APOD) dataset. This project will give you hands-on experience with key skills like:

• Data Cleaning Pipelines: Prepare and clean the dataset for optimal search performance.

• Tokenization and Analysis: Break down text data to enable efficient search queries.

• Search Functionality: Implement powerful search features, including pagination and filtering.

• Aggregations: Summarize and analyze search results to extract insights.

By the end of the project, you'll have a fully functional search engine and a deeper understanding of how Elasticsearch can enhance your applications.

Course Contents

This in-depth course spans 5 hours and covers a wide range of topics:

1. Introduction and Installation

2. Index Management and Document Storage

3. Text Analysis, Searching, and Pipelines

4. Advanced Features like Embeddings and Semantic Search

5. Final Project – Building a Real-World Search Engine

Why Learn Elasticsearch?

Elasticsearch is a powerful tool used by companies worldwide for search, logging, and analytics. Whether you're a developer, data scientist, or tech enthusiast, mastering Elasticsearch can open new career opportunities and enhance your ability to build efficient, scalable applications. This course makes learning Elasticsearch accessible, practical, and fun!

You can watch the full course on the freeCodeCamp.org YouTube channel (5-hour watch).

Location-based features are pretty common in apps nowadays. These features might seem complicated, but they can actually be implemented quite easily with Elasticsearch.

Elasticsearch is a NoSQL database with a document-based structure. It's often used as a Search Engine. It also provides its own syntax and many tools to help your search be as flexible as possible.

In this article I will show you a simple way to search by geolocation by getting a list of cities by coordinate range.

How to Install Elasticsearch

You can find an easy-to-follow installation guide on Elasticsearch's website. At the time I am writing this article, I am using Elasticsearch version 7.4.2 .

Just keep in mind that Elasticsearch has made a lot of changes in recent versions, one of them being the removal of mapping types. So if you are using another version of Elasticsearch some things here might not fully work.

After finishing your installation, do not forget to run your Elasticsearch service, which they emphasize clearly in their installation guide (for Linux, do this ./bin/elasticsearch ).

Make sure your elasticsearch is running by using a GET request into port 9200 in your local machine, like this: GET http://localhost:9200

How to Make Your Elasticsearch Index

An index is similar to table in a regular database. For this example, let's make an index named cities that will contain our data.

Let's also define a simple model for our data:

• id : keyword for our identifier
• name : text for the city name
• coordinate : geo_point to store our city coordinates (neat, they have this data-type already)

In Elasticsearch, we create the index by making a curl into an API. In our case our request will be like this:

PUT http://localhost:9200/cities

{
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 1
    },
    "mappings": {
        "properties": {
            "id": {
                "type": "keyword"
            },
            "name": {
                "type": "text"
            },
            "coordinate": {
                "type": "geo_point"
            }
        }
    }
}

When you used that curl, you should get a response like this to verify that your index has been created:

{
    "acknowledged": true,
    "shards_acknowledged": true,
    "index": "cities"
}

Nicely done! Now your index is ready to be used. Let's go ahead and play around with our newly created index.

How to Populate Elasticsearch Data

We will now fill our Elasticsearch index with documents. If you are not familiar with this term, just know that it is very similar to rows in a SQL database.

In Elasticsearch, it's possible to store data that doesn't match with our predefined schema. But we will not do that here – instead we will insert data that matches our predefined schema.

Since we will be inserting multiple data at once, we will use the bulk API that Elasticsearch provides that allows multiple insertions in one API call.

In the example below, I will be inserting 9 cities into my index. Feel free to add more if you wish.

POST '[http://localhost:9200/cities/_bu](http://localhost:9200/cities/_bu)lk

{ "index":{"_index": "cities" } }
{ "id": 1, "name": "Jakarta", "coordinate": {  "lat": -6.2008, "lon": 106.8456}}
{ "index":{"_index": "cities" } }
{ "id": 2, "name": "Tokyo", "coordinate": {  "lat": 35.6762, "lon": 139.6503} }
{ "index":{"_index": "cities" } }
{ "id": 3, "name": "Hong Kong", "coordinate": {  "lat": 22.3193, "lon": 114.1694} }
{ "index":{"_index": "cities" } }
{ "id": 4, "name": "New York", "coordinate": {  "lat": 40.7128, "lon": -74.0060} }
{ "index":{"_index": "cities" } }
{ "id": 5, "name": "Paris", "coordinate": {  "lat": 48.8566, "lon": 2.3522} }
{ "index":{"_index": "cities" } }
{ "id": 6, "name": "Bali", "coordinate": {  "lat": -8.3405, "lon": 115.0920} }
{ "index":{"_index": "cities" } }
{ "id": 7, "name": "Berlin", "coordinate": {  "lat": 52.5200, "lon": 13.4050} }
{ "index":{"_index": "cities" } }
{ "id": 8, "name": "San Fransisco", "coordinate": {  "lat": 37.7749, "lon": -122.4194} }
{ "index":{"_index": "cities" } }
{ "id": 9, "name": "Beijing", "coordinate": {  "lat": 39.9042, "lon": 166.4074} }

The payload might looks weird since it's in an incorrect JSON format, but don't worry – it's supposedly designed that way.

It should then reply back to you with a response similar to this:

{
    "took": 72,
    "errors": false,
    "items": [
        //will contains item for each data inserted
        ...
    ]
}

How to Query Your Elasticsearch Documents

_Photo by [Unsplash](https://unsplash.com/@chrislawton?utm_source=ghost&utm_medium=referral&utm_campaign=api-credit">Chris Lawton / <a href="https://unsplash.com/?utm_source=ghost&utm_medium=referral&utmcampaign=api-credit)

Now comes the interesting part. We are going to do some querying with the documents that we inserted before.

Elasticsearch supports many types of syntax for query searching. It also has geolocation type searching which we will play around with today.

We can simply start searching for our cities with curl like this:

POST '[http://localhost:9200/cities/_sear](http://localhost:9200/cities/_sear)ch

{
  "query": {
    "bool": {
      "filter": {
        "geo_distance": {
          "distance": "10km",
          "coordinate": {
            "lat": 37.76,
            "lon": -122.42
          }
        }
      }
    }
  }
}

That query should give me San Francisco, and the coordinates 37.7749 and -122.4194 should be inside a 10km distance radius from our coordinates (courtesy of Google).

{
    "took": 7,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 1,
            "relation": "eq"
        },
        "max_score": 0.0,
        "hits": [
            {
                "_index": "cities",
                "_type": "_doc",
                "_id": "eKPspHYBivyIhfWHb2vl",
                "_score": 0.0,
                "_source": {
                    "id": 8,
                    "name": "San Fransisco",
                    "coordinate": {
                        "lat": 37.7749,
                        "lon": -122.4194
                    }
                }
            }
        ]
    }
}

Congratulations! Now you have your own location search engine.
But let's experiment a bit more. Let's say you want to get more cities in that location.

Let's try to expand the distance to 4500km by changing the payload:

{
  "query": {
    "bool": {
      "filter": {
        "geo_distance": {
          "distance": "4500km",
          "coordinate": {
            "lat": 37.76,
            "lon": -122.42
          }
        }
      }
    }
  }
}

And you should get this response:

{
    "took": 8,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 2,
            "relation": "eq"
        },
        "max_score": 0.0,
        "hits": [
            {
                "_index": "cities",
                "_type": "_doc",
                "_id": "dKPspHYBivyIhfWHb2vl",
                "_score": 0.0,
                "_source": {
                    "id": 4,
                    "name": "New York",
                    "coordinate": {
                        "lat": 40.7128,
                        "lon": -74.0060
                    }
                }
            },
            {
                "_index": "cities",
                "_type": "_doc",
                "_id": "eKPspHYBivyIhfWHb2vl",
                "_score": 0.0,
                "_source": {
                    "id": 8,
                    "name": "San Fransisco",
                    "coordinate": {
                        "lat": 37.7749,
                        "lon": -122.4194
                    }
                }
            }
        ]
    }
}

It gives two results: New York and San Fransisco. The results look correct, but the positioning might be a bit weird. San Fransisco supposedly should come first since it's closer, right?

Well not exactly, since what we are doing is just filtering. Our query is just filtering and it doesn't care about which one is closest to you.

But what if we want to do some calculation to show which locations might be the nearest? Don't worry, Elasticsearch can do that too. We can use a type of query called a function score query.

How to Use a Function Score Query in Elasticsearch

Elasticsearch calculates (scores) what documents it will show to the user. By using function score queries we can modify that score so we can determine which documents should be returned.

Here, we will be using the decay query function. There are three kinds of decay functions: exp, linear, and gauss. Each of them has different behaviors.

Image source

The one we will use here is the linear type function. We will also specify the coordinates together with offset and scale.

POST '[http://localhost:9200/cities/_sear](http://localhost:9200/cities/_sear)ch

{
  "query": {
    "function_score": {
      "functions": [
        {
          "linear": {
            "coordinate": {
              "origin": "37, -122",
              "offset": "100km",
              "scale":"2500km"
            }
          }
        }
      ],
       "min_score":"0.1"
    }
  }
}

Now, we should get our results ordered by the highest score.

{
    "took": 32,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 2,
            "relation": "eq"
        },
        "max_score": 1.0,
        "hits": [
            {
                "_index": "cities",
                "_type": "_doc",
                "_id": "eKPspHYBivyIhfWHb2vl",
                "_score": 1.0,
                "_source": {
                    "id": 8,
                    "name": "San Fransisco",
                    "coordinate": {
                        "lat": 37.7749,
                        "lon": -122.4194
                    }
                }
            },
            {
                "_index": "cities",
                "_type": "_doc",
                "_id": "dKPspHYBivyIhfWHb2vl",
                "_score": 0.19508117,
                "_source": {
                    "id": 4,
                    "name": "New York",
                    "coordinate": {
                        "lat": 40.7128,
                        "lon": -74.0060
                    }
                }
            }
        ]
    }
}

And that wraps it up!

Conclusion

In this article, we've covered how to implement location-based search with Elasticsearch. But this is not the end – what I have shown here is just the surface of what you can do.

I hope you found this article interesting and useful. If so, keep learning more about it and try to experiment with combining function scoring. It will be fun, I promise.

Always be curios and you will learn something new.

Change Data Capture (CDC) involves observing the changes happening in a database and making them available in a form that can be exploited by other systems.

One of the most interesting use-cases is to make them available as a stream of events. This means you can, for example, catch the events and update a search index as the data are written to the database.

Interesting right? Let's see how to implement a CDC system that can observe the changes made to a NoSQL database (MongoDB), stream them through a message broker (Kafka), process the messages of the stream (Kafka Streams), and update a search index (Elasticsearch)!?

TL;DR

The full code of the project is available on GitHub in this repository. If you want to skip all my jibber jabber and just run the example, go straight to the How to run the project section near the end of the article!?

Use case & infrastructure

We run a web application that stores photos uploaded by users. People can share their shots, let others download them, create albums, and so on. Users can also provide a description of their photos, as well as Exif metadata and other useful information.

We want to store such information and use it to improve our search engine. We will focus on this part of our system that is depicted in the following diagram.

Photo information storage architecture

The information is provided in JSON format. Since I like to post my shots on Unsplash, and the website provides free access to its API, I used their model for the photo JSON document.

Once the JSON is sent through a POST request to our server, we store the document inside a MongoDB database. We will also store it in Elasticsearch for indexing and quick search.

However, we love long exposure shots, and we would like to store in a separate index a subset of information regarding this kind of photo. It can be the exposure time, as well as the location (latitude and longitude) where the photo has been taken. In this way, we can create a map of locations where photographers usually take long exposure photos.

Here comes the interesting part: instead of explicitly calling Elasticsearch in our code once the photo info is stored in MongoDB, we can implement a CDC exploiting Kafka and Kafka Streams.

We listen to modifications to MongoDB oplog using the interface provided by MongoDB itself. When the photo is stored we send it to a photo Kafka topic. Using Kafka Connect, an Elasticsearch sink is configured to save everything sent to that topic to a specific index. In this way, we can index all photos stored in MongoDB automatically.

We need to take care of the long exposure photos too. It requires some processing of the information to extract what we need. For this reason, we use Kafka Streams to create a processing topology to:

1. Read from the photo topic
2. Extract Exif and location information
3. Filter long exposure photos (exposure time > 1 sec.)
4. Write to a long-exposure topic.

Then another Elasticsearch sink will read data from the long-exposure topic and write it to a specific index in Elasticsearch.

It is quite simple, but it's enough to have fun with CDC and Kafka Streams! ?

Server implementation

Let's have a look at what we need to implement: our server exposing the REST APIs!

Models and DAO

First things first, we need a model of our data and a Data Access Object (DAO) to talk to our MongoDB database.

As I said, the model for the photo JSON information is the one used by Unsplash. Check out the free API documentation for an example of the JSON we will use.

I created the mapping for the serializaion/deserialization of the photo JSON using spray-json. I'll skip the details about this, if you are curious just look at the repo!

Let's focus on the model for the long exposusure photo.

case class LongExposurePhoto(id: String, exposureTime: Float, createdAt: Date, location: Location)

object LongExposurePhotoJsonProtocol extends DefaultJsonProtocol {
  implicit val longExposurePhotoFormat:RootJsonFormat[LongExposurePhoto] = jsonFormat(LongExposurePhoto, "id", "exposure_time", "created_at", "location")
}

This is quite simple: we keep from the photo JSON the information about the id, the exposure time (exposureTime), when the photo has been created (createdAt), and the location where it has been taken. The location comprehends the city, the country, and the position composed of latitude and longitude.

The DAO consists of just the PhotoDao.scala class.

class PhotoDao(database: MongoDatabase, photoCollection: String) {

  val collection: MongoCollection[Document] = database.getCollection(photoCollection)

  def createPhoto(photo: Photo): Future[String] = {
    val doc = Document(photo.toJson.toString())
    doc.put("_id", photo.id)
    collection.insertOne(doc).toFuture()
      .map(_ => photo.id)
  }
}

Since I want to keep this example minimal and focused on the CDC implementation, the DAO has just one method to create a new photo document in MongoDB.

It is straightforward: create a document from the photo JSON, and insert it in mongo using id as the one of the photo itself. Then, we can return the id of the photo just inserted in a Future (the MongoDB API is async).

Kafka Producer

Once the photo is stored inside MongoDB, we have to send it to the photo Kafka topic. This means we need a producer to write the message in its topic. The PhotoProducer.scala class looks like this.

case class PhotoProducer(props: Properties, topic: String) {

  createKafkaTopic(props, topic)
  val photoProducer = new KafkaProducer[String, String](props)

  def sendPhoto(photo: Photo): Future[RecordMetadata] = {
    val record = new ProducerRecord[String, String](topic, photo.id, photo.toJson.compactPrint)
    photoProducer.send(record)
  }

  def closePhotoProducer(): Unit = photoProducer.close()
}

I would say that this is pretty self-explanatory. The most interesting part is probably the createKafkaTopic method that is implemented in the utils package.

def createKafkaTopic(props: Properties, topic: String): Unit = {
    val adminClient = AdminClient.create(props)
    val photoTopic = new NewTopic(topic, 1, 1)
    adminClient.createTopics(List(photoTopic).asJava)
  }

This method creates the topic in Kafka setting 1 as a partition and replication factor (it is enough for this example). It is not required, but creating the topic in advance lets Kafka balance partitions, select leaders, and so on. This will be useful to get our stream topology ready to process as we start our server.

Event Listener

We have the DAO that writes in MongoDB and the producer that sends the message in Kafka. We need to glue them together in some way so that when the document is stored in MongoDB the message is sent to the photo topic. This is the purpose of the PhotoListener.scala class.

case class PhotoListener(collection: MongoCollection[Document], producer: PhotoProducer) {

  val cursor: ChangeStreamObservable[Document] = collection.watch()

  cursor.subscribe(new Observer[ChangeStreamDocument[Document]] {
    override def onNext(result: ChangeStreamDocument[Document]): Unit = {
      result.getOperationType match {
        case OperationType.INSERT => {
          val photo = result.getFullDocument.toJson().parseJson.convertTo[Photo]
          producer.sendPhoto(photo).get()
          println(s"Sent photo with Id ${photo.id}")
        }
        case _ => println(s"Operation ${result.getOperationType} not supported")
      }
    }
    override def onError(e: Throwable): Unit = println(s"onError: $e")
    override def onComplete(): Unit = println("onComplete")})
}

We exploit the Chage Streams interface provided by the MongoDB scala library.

Here is how it works: we watch() the collection where photos are stored. When there is a new event (onNext) we run our logic.

For this example we are interested only in the creation of new documents, so we explicitly check that the operation is of type OperationType.INSERT. If the operation is the one we are interested in, we get the document and convert it to a Photo object to be sent by our producer.

That's it! With few lines of code we connected the creation of documents in MongoDB to a stream of events in Kafka.?

As a side note, be aware that to use the Change Streams interface we have to setup a MongoDB replica set. This means we need to run 3 instances of MongoDB and configure them to act as a replica set using the following command in mongo client:

rs.initiate({_id : "r0", members: [{ _id : 0, host : "mongo1:27017", priority : 1 },{ _id : 1, host :"mongo2:27017", priority : 0 },{ _id : 2, host : "mongo3:27017", priority : 0, arbiterOnly: true }]})

Here our instances are the containers we will run in the docker-compose file, that is mongo1, mongo2, and mongo3.

Processing Topology

Time to build our processing topology! It will be in charge of the creation of the long-exposure index in Elasticsearch. The topology is described by the following diagram:

Processing topology

and it is implemented in the LongExposureTopology.scala object class. Let's analyse every step of our processing topology.

val stringSerde = new StringSerde

val streamsBuilder = new StreamsBuilder()

val photoSource: KStream[String, String] = streamsBuilder.stream(sourceTopic, Consumed.`with`(stringSerde, stringSerde))

The first step is to read from a source topic. We start a stream from the sourceTopic (that is photo topic) using the StreamsBuilder() object. The stringSerde object is used to serialise and deserialise the content of the topic as a String.

Please notice that at each step of the processing we create a new stream of data with a KStream object. When creating the stream, we specify the key and the value produced by the stream. In our topology the key will always be a String. In this step the value produced is still a String.

val covertToPhotoObject: KStream[String, Photo] =
      photoSource.mapValues((_, jsonString) => {
        val photo = jsonString.parseJson.convertTo[Photo]
        println(s"Processing photo ${photo.id}")
        photo
      })

The next step is to convert the value extracted from the photo topic into a proper Photo object.

So we start from the photoSource stream and work on the values using the mapValues function. We simply parse the value as a JSON and create the Photo object that will be sent in the convertToPhotoObject stream.

val filterWithLocation: KStream[String, Photo] = covertToPhotoObject.filter((_, photo) => photo.location.exists(_.position.isDefined))

There is no guarantee that the photo we are processing will have the info about the location, but we want it in our long exposure object. This step of the topology filters out from the covertToPhotoObject stream the photos that have no info about the location, and creates the filterWithLocation stream.

val filterWithExposureTime: KStream[String, Photo] = filterWithLocation.filter((_, photo) => photo.exif.exists(_.exposureTime.isDefined))

Another important fact for our processing is the exposure time of the photo. For this reason, we filter out from the filterWithLocation stream the photos without exposure time info, creating the filterWithExposureTime.

val dataExtractor: KStream[String, LongExposurePhoto] =
      filterWithExposureTime.mapValues((_, photo) => LongExposurePhoto(photo.id, parseExposureTime(photo.exif.get.exposureTime.get), photo.createdAt, photo.location.get))

We now have all we need to create a LongExposurePhoto object! That is the result of the dataExtractor: it takes the Photo coming from the filterWithExposureTime stream and produces a new stream containing LongExposurePhoto.

val longExposureFilter: KStream[String, String] =
      dataExtractor.filter((_, item) => item.exposureTime > 1.0).mapValues((_, longExposurePhoto) => {
        val jsonString = longExposurePhoto.toJson.compactPrint
        println(s"completed processing: $jsonString")
        jsonString
      })

We are almost there. We now have to keep the photos with a long exposure time (that we decided is more then 1 sec.). So we create a new longExposureFilter stream without the photos that are not long exposure.

This time we also serialise the LongExposurePhotos into the corresponding JSON string, which will be written to Elasticsearch in the next step.

longExposureFilter.to(sinkTopic, Produced.`with`(stringSerde, stringSerde))

streamsBuilder.build()

This is the last step of our topology. We write to our sinkTopic (that is long-exposure topic) using the string serialiser/deserialiser what is inside the longExposureFilter stream. The last command simply builds the topology we just created.

Now that we have our topology, we can use it in our server. The PhotoStreamProcessor.scala class is what manages the processing.

class PhotoStreamProcessor(kafkaProps: Properties, streamProps: Properties, sourceTopic: String, sinkTopic: String) {

  createKafkaTopic(kafkaProps, sinkTopic)
  val topology: Topology = LongExposureTopology.build(sourceTopic, sinkTopic)
  val streams: KafkaStreams = new KafkaStreams(topology, streamProps)

  sys.ShutdownHookThread {
    streams.close(java.time.Duration.ofSeconds(10))
  }

  def start(): Unit = new Thread {
    override def run(): Unit = {
      streams.cleanUp()
      streams.start()
      println("Started long exposure processor")
    }
  }.start()
}

First we create the sinkTopic, using the same utility method we saw before. Then we build the stream topology and initialize a KafkaStreams object with that topology.

To start the stream processing, we need to create a dedicated Thread that will run the streaming while the server is alive. According to the official documentation, it is always a good idea to cleanUp() the stream before starting it.

Our PhotoStreamProcessor is ready to go!?

REST API

The server exposes REST APIs to send it the photo information to store. We make use of Akka HTTP for the API implementation.

trait AppRoutes extends SprayJsonSupport {

  implicit def system: ActorSystem
  implicit def photoDao: PhotoDao
  implicit lazy val timeout = Timeout(5.seconds)

  lazy val healthRoute: Route = pathPrefix("health") {
    concat(
      pathEnd {
        concat(
          get {
            complete(StatusCodes.OK)
          }
        )
      }
    )
  }

  lazy val crudRoute: Route = pathPrefix("photo") {
    concat(
      pathEnd {
        concat(
          post {
            entity(as[Photo]) { photo =>
              val photoCreated: Future[String] =
                photoDao.createPhoto(photo)
              onSuccess(photoCreated) { id =>
              complete((StatusCodes.Created, id))
              }
            }
          }
        )
      }
    )
  }

}

To keep the example minimal, we have only two routes:

• GET /health - to check if the server is up & running
• POST /photo - to send to the system the JSON of the photo information we want to store. This endpoint uses the DAO to store the document in MongoDB and returns a 201 with the id of the stored photo if the operation succeeded.

This is by no means a complete set of APIs, but it is enough to run our example.?

Server main class

OK, we implemented all the components of our server, so it's time to wrap everything up. This is our Server.scala object class.

implicit val system: ActorSystem = ActorSystem("kafka-stream-playground")
implicit val materializer: ActorMaterializer = ActorMaterializer()

First a couple of Akka utility values. Since we use Akka HTTP to run our server and REST API, these implicit values are required.

val config: Config = ConfigFactory.load()
val address = config.getString("http.ip")
val port = config.getInt("http.port")

val mongoUri = config.getString("mongo.uri")
val mongoDb = config.getString("mongo.db")
val mongoUser = config.getString("mongo.user")
val mongoPwd = config.getString("mongo.pwd")
val photoCollection = config.getString("mongo.photo_collection")

val kafkaHosts = config.getString("kafka.hosts").split(',').toList
val photoTopic = config.getString("kafka.photo_topic")
val longExposureTopic = config.getString("kafka.long_exposure_topic")

Then we read all the configuration properties. We will come back to the configuration file in a moment.

val kafkaProps = new Properties()
kafkaProps.put("bootstrap.servers", kafkaHosts.mkString(","))
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

val streamProps = new Properties()
streamProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "long-exp-proc-app")
streamProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaHosts.mkString(","))

val photoProducer = PhotoProducer(kafkaProps, photoTopic)
val photoStreamProcessor = new PhotoStreamProcessor(kafkaProps, streamProps, photoTopic, "long-exposure")
photoStreamProcessor.start()

We have to configure both our Kafka producer and the stream processor. We also start the stream processor, so the server will be ready to process the documents sent to it.

val client = MongoClient(s"mongodb://$mongoUri/$mongoUser")
val db = client.getDatabase(mongoDb)
val photoDao: PhotoDao = new PhotoDao(db, photoCollection)
val photoListener = PhotoListener(photoDao.collection, photoProducer)

Also MongoDB needs to be configured. We setup the connection and initialize the DAO as well as the listener.

lazy val routes: Route = healthRoute ~ crudRoute

Http().bindAndHandle(routes, address, port)
Await.result(system.whenTerminated, Duration.Inf)

Everything has been initialized. We create the REST routes for the communication to the server, bind them to the handlers, and finally start the server!?

Server configuration

This is the configuration file used to setup the server:

http {
  ip = "127.0.0.1"
  ip = ${?SERVER_IP}

  port = 8000
  port = ${?SERVER_PORT}
}
mongo {
  uri = "127.0.0.1:27017"
  uri = ${?MONGO_URI}
  db = "kafka-stream-playground"
  user = "admin"
  pwd = "admin"
  photo_collection = "photo"
}
kafka {
  hosts = "127.0.0.1:9092"
  hosts = ${?KAFKA_HOSTS}
  photo_topic = "photo"
  long_exposure_topic = "long-exposure"
}

I think that this one does not require much explanation, right??

Connectors configuration

The server we implemented writes in two Kafka topics: photo and long-exposure. But how are messages written in Elasticsearch as documents? Using Kafka Connect!

We can setup two connectors, one per topic, and tell the connectors to write every message going through that topic in Elasticsearch.

First we need Kafka Connect. We can use the container provided by Confluence in the docker-compose file:

connect:
    image: confluentinc/cp-kafka-connect
    ports:
      - 8083:8083
    networks:
      - kakfa_stream_playground
    depends_on:
      - zookeeper
      - kafka
    volumes:
      - $PWD/connect-plugins:/connect-plugins
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka:9092
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.storage.StringConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_ZOOKEEPER_CONNECT: zookeeper:2181
      CONNECT_PLUGIN_PATH: /connect-plugins
      CONNECT_LOG4J_ROOT_LOGLEVEL: INFO

I want to focus on some of the configuration values.

First of all, we need to expose the port 8083 - that will be our endpoint to configure the connectors (CONNECT_REST_PORT).

We also need to map a volume to the /connect-plugins path, where we will place the Elasticsearch Sink Connector to write to Elasticsearch. This is reflected also in the CONNECT_PLUGIN_PATH.

The connect container should know how to find the Kafka servers, so we set CONNECT_BOOTSTRAP_SERVERS as kafka:9092.

Once Kafka Connect is ready, we can send the configurations of our connectors to the http://localhost:8083/connectors endpoint. We need 2 connectors, one for the photo topic and one for the long-exposure topic. We can send the configuration as a JSON with a POST request.

{
  "name": "photo-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "photo",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "schema.ignore": "true",
    "connection.url": "http://elastic:9200",
    "type.name": "kafka-connect",
    "behavior.on.malformed.documents": "warn",
    "name": "photo-connector"
  }
}

We explicitly say we are gonna use the ElasticsearchSinkConnector as the connector.class , as well as the topics that we want to sink - in this case photo.

We don't want to use a schema for the value.converter, so we can disable it (value.converter.schemas.enable) and tell the connector to ignore the schema (schema.ignore).

The connector for the long-exposure topic is exactly like this one. The only difference is the name and of course the topics.

How to run the project

We have all we need to test the CDC! How can we do it? It's quite easy: simply run the setup.sh script in the root folder of the repo!

What will the script do?

1. Run the docker-compose file with all the services.
2. Configure MongoDB replica set. This is required to enable the Change Stream interface to capture data changes. More info about this here.
3. Configure the Kafka connectors.
4. Connect to the logs of the server.

The docker-compose will run the following services:

• Our Server
• 3 instances of MongoDB (required for the replica set)
• Mongoku, a MongoDB client
• Kafka (single node)
• Kafka connect
• Zookeeper (required by Kafka)
• Elasticsearch
• Kibana

There are a lot of containers to run, so make sure you have enough resources to run everything properly. If you want, remove Mongoku and Kibana from the compose-file, since they are used just for a quick look inside the DBs.

Once everything is up and running, you just have to send data to the server.

I collected some JSON documents of photos from Unplash that you can use to test the system in the photos.txt file.

There are a total of 10 documents, with 5 of them containing info about long exposure photos. Send them to the server running the send-photos.sh script in the root of the repo. Check that everything is stored in MongoDB connecting to Mongoku at http://localhost:3100. Then connect to Kibana at http://localhost:5601 and you will find two indexes in Elasticsearch: photo, containing the JSON of all the photos stored in MongoDB, and long-exposure, containing just the info of the long exposure photos.

Amazing, right? ?

Conclusion

We made it guys!?

Starting from the design of the use-case, we built our system that connected a MongoDB database to Elasticsearch using CDC.

Kafka Streams is the enabler, allowing us to convert database events to a stream that we can process.

Do you need to see the whole project? Just checkout the repository on GitHub!?

That's it, enjoy! ?

My last task at BigPanda was to upgrade an existing service that was using Elasticsearch version 1.7 to a newer Elasticsearch version, 6.8.1.

In this post, I will share how we migrated from Elasticsearch 1.6 to 6.8 with harsh constraints like zero downtime, no data loss, and zero bugs. I'll also provide you with a script that does the migration for you.

This post contains 6 chapters (and one is optional):

• What’s in it for me? --> What were the new features that led us to upgrade our version?
• The constraints --> What were our business requirements?
• Problem solving --> How did we address the constraints?
• Moving forward --> The plan.
• [Optional chapter] --> How did we handle the infamous mapping explosion problem?
• Finally --> How to do data migration between clusters.

Chapter 1 — What’s in it for me?

What benefits were we expecting to solve by upgrading our data store?

There were a couple of reasons:

1. Performance and stability issues — We were experiencing a huge number of outages with long MTTR that caused us a lot of headaches. This was reflected in frequent high latencies, high CPU usage, and more issues.
2. Non-existent support in old Elasticsearch versions — We were missing some operative knowledge in Elasticsearch, and when we searched for outside consulting we were encouraged to migrate forward to receive support.
3. Dynamic mappings in our schema — Our current schema in Elasticsearch 1.7 used a feature called dynamic mappings that made our cluster explode multiple times. So we wanted to address this issue.
4. Poor visibility on our existing cluster — We wanted a better view under the hood and saw that later versions had great metrics exporting tools.

Chapter 2 — The constraints

• ZERO downtime migration — We have active users on our system, and we could not afford for the system to be down while we were migrating.
• Recovery plan — We could not afford to “lose” or “corrupt” data, no matter the cost. So we needed to prepare a recovery plan in case our migration failed.
• Zero bugs — We could not change existing search functionality for end-users.

Chapter 3 — Problem solving and thinking of a plan

Let’s tackle the constraints from the simplest to the most difficult:

Zero bugs

In order to address this requirement, I studied all the possible requests the service receives and what its outputs were. Then I added unit-tests where needed.

In addition, I added multiple metrics (to the Elasticsearch Indexer and the new Elasticsearch Indexer ) to track latency, throughput, and performance, which allowed me to validate that we only improved them.

Recovery plan

This means that I needed to address the following situation: I deployed the new code to production and stuff was not working as expected. What can I do about it then

Since I was working in a service that used event-sourcing, I could add another listener (diagram attached below) and start writing to a new Elasticsearch cluster without affecting production status

Zero downtime migration

The current service is in live mode and cannot be “deactivated” for periods longer than 5–10 minutes. The trick to getting this right is this:

• Store a log of all the actions your service is handling (we use Kafka in production)
• Start the migration process offline (and keep track of the offset before you started the migration)
• When the migration ends, start the new service against the log with the recorded offset and catch up the lag
• When the lag finishes, change your frontend to query against the new service and you are done

Chapter 4 — The plan

Our current service uses the following architecture (based on message passing in Kafka):

1. Event topic contains events produced by other applications (for example, UserId 3 created)
2. Command topic contains the translation of these events into specific commands used by this application (for example: Add userId 3)
3. Elasticsearch 1.7 — The datastore of the command Topic read by the Elasticsearch Indexer.

We planned to add another consumer (new Elasticsearch Indexer) to the command topic, which will read the same exact messages and write them in parallel to Elasticsearch 6.8.

Where should I start?

To be honest, I considered myself a newbie Elasticsearch user. To feel confident to perform this task, I had to think about the best way to approach this topic and learn it. A few things that helped were:

1. Documentation — It’s an insanely useful resource for everything Elasticsearch. Take the time to read it and take notes (don’t miss: Mapping and QueryDsl).
2. HTTP API — everything under CAT API. This was super useful to debug things locally and see how Elasticsearch responds (don’t miss: cluster health, cat indices, search, delete index).
3. Metrics (❤️) — From the first day, we configured a shiny new dashboard with lots of cool metrics (taken from elasticsearch-exporter-for-Prometheus) that helped and pushed us to understand more about Elasticsearch.

The code

Our codebase was using a library called elastic4s and was using the oldest release available in the library — a really good reason to migrate! So the first thing to do was just to migrate versions and see what broke.

There are a few tactics on how to do this code migration. The tactic we chose was to try and restore existing functionality first in the new Elasticsearch version without re-writing the all code from the start. In other words, to reach existing functionality but on a newer version of Elasticsearch.

Luckily for us, the code already contained almost full testing coverage so our task was much much simpler, and that took around 2 weeks of development time.

It's important to note that, if that wasn't the case, we would have had to invest some time in filling that coverage up. Only then would we be able to migrate since one of our constraints was to not break existing functionality.

Chapter 5 — The mapping explosion problem

Let’s describe our use-case in more detail. This is our model:

class InsertMessageCommand(tags: Map[String,String])

And for example, an instance of this message would be:

new InsertMessageCommand(Map("name"->"dor","lastName"->"sever"))

And given this model, we needed to support the following query requirements:

1. Query by value
2. Query by tag name and value

The way this was modeled in our Elasticsearch 1.7 schema was using a dynamic template schema (since the tag keys are dynamic, and cannot be modeled in advanced).

The dynamic template caused us multiple outages due to the mapping explosion problem, and the schema looked like this:

curl -X PUT "localhost:9200/_template/my_template?pretty" -H 'Content-Type: application/json' -d '
{
    "index_patterns": [
        "your-index-names*"
    ],
    "mappings": {
            "_doc": {
                "dynamic_templates": [
                    {
                        "tags": {
                            "mapping": {
                                "type": "text"
                            },
                            "path_match": "actions.tags.*"
                        }
                    }
                ]
            }
        },
    "aliases": {}
}'  

curl -X PUT "localhost:9200/your-index-names-1/_doc/1?pretty" -H 'Content-Type: application/json' -d'
{
  "actions": {
    "tags" : {
        "name": "John",
        "lname" : "Smith"
    }
  }
}
'

curl -X PUT "localhost:9200/your-index-names-1/_doc/2?pretty" -H 'Content-Type: application/json' -d'
{
  "actions": {
    "tags" : {
        "name": "Dor",
        "lname" : "Sever"
  }
}
}
'

curl -X PUT "localhost:9200/your-index-names-1/_doc/3?pretty" -H 'Content-Type: application/json' -d'
{
  "actions": {
    "tags" : {
        "name": "AnotherName",
        "lname" : "AnotherLastName"
  }
}
}
'

curl -X GET "localhost:9200/_search?pretty" -H 'Content-Type: application/json' -d'
{
    "query": {
        "match" : {
            "actions.tags.name" : {
                "query" : "John"
            }
        }
    }
}
'
# returns 1 match(doc 1)


curl -X GET "localhost:9200/_search?pretty" -H 'Content-Type: application/json' -d'
{
    "query": {
        "match" : {
            "actions.tags.lname" : {
                "query" : "John"
            }
        }
    }
}
'
# returns zero matches

# search by value
curl -X GET "localhost:9200/_search?pretty" -H 'Content-Type: application/json' -d'
{
    "query": {
        "query_string" : {
            "fields": ["actions.tags.*" ],
            "query" : "Dor"
        }
    }
}
'

Nested documents solution

Our first instinct in solving the mapping explosion problem was to use nested documents.

We read the nested data type tutorial in the Elastic docs and defined the following schema and queries:

curl -X PUT "localhost:9200/my_index?pretty" -H 'Content-Type: application/json' -d'
{
        "mappings": {
            "_doc": {
            "properties": {
            "tags": {
                "type": "nested" 
                }                
            }
        }
        }
}
'

curl -X PUT "localhost:9200/my_index/_doc/1?pretty" -H 'Content-Type: application/json' -d'
{
  "tags" : [
    {
      "key" : "John",
      "value" :  "Smith"
    },
    {
      "key" : "Alice",
      "value" :  "White"
    }
  ]
}
'


# Query by tag key and value
curl -X GET "localhost:9200/my_index/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "query": {
    "nested": {
      "path": "tags",
      "query": {
        "bool": {
          "must": [
            { "match": { "tags.key": "Alice" }},
            { "match": { "tags.value":  "White" }} 
          ]
        }
      }
    }
  }
}
'

# Returns 1 document


curl -X GET "localhost:9200/my_index/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "query": {
    "nested": {
      "path": "tags",
      "query": {
        "bool": {
          "must": [
            { "match": { "tags.value":  "Smith" }} 
          ]
        }
      }
    }
  }
}
'

# Query by tag value
# Returns 1 result

And this solution worked. However, when we tried to insert real customer data we saw that the number of documents in our index increased by around 500 times.

We thought about the following problems and went on to find a better solution:

1. The amount of documents we had in our cluster was around 500 million documents. This meant that, with the new schema, we were going to reach two hundred fifty billion documents (that’s 250,000,000,000 documents ?).
2. We read this really good blog post — https://blog.gojekengineering.com/elasticsearch-the-trouble-with-nested-documents-e97b33b46194 which highlights that nested documents can cause high latency in queries and heap usage problems.
3. Testing — Since we were converting 1 document in the old cluster to an unknown number of documents in the new cluster, it would be much harder to track if the migration process worked without any data loss. If our conversion was 1:1, we could assert that the count in the old cluster equalled the count in the new cluster.

Avoiding nested documents

The real trick in this was to focus on what supported queries we were running: search by tag value, and search by tag key and value.

The first query does not require nested documents since it works on a single field. For the latter, we did the following trick. We created a field that contains the combination of the key and the value. Whenever a user queries on a key, value match, we translate their request to the corresponding text and query against that field.

Example:

curl -X PUT "localhost:9200/my_index_2?pretty" -H 'Content-Type: application/json' -d'
{
    "mappings": {
        "_doc": {
            "properties": {
                "tags": {
                    "type": "object",
                    "properties": {
                        "keyToValue": {
                            "type": "keyword"
                        },
                        "value": {
                            "type": "keyword"
                        }
                    }
                }
            }
        }
    }
}
'


curl -X PUT "localhost:9200/my_index_2/_doc/1?pretty" -H 'Content-Type: application/json' -d'
{
  "tags" : [
    {
      "keyToValue" : "John:Smith",
      "value" : "Smith"
    },
    {
      "keyToValue" : "Alice:White",
      "value" : "White"
    }
  ]
}
'

# Query by key,value
# User queries for key: Alice, and value : White , we then query elastic with this query:

curl -X GET "localhost:9200/my_index_2/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "query": {
        "bool": {
          "must": [ { "match": { "tags.keyToValue": "Alice:White" }}]
  }}}
'

# Query by value only
curl -X GET "localhost:9200/my_index_2/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "query": {
        "bool": {
          "must": [ { "match": { "tags.value": "White" }}]
  }}}
'

Chapter 6 — The migration process

We planned to migrate about 500 million documents with zero downtime. To do that we needed:

1. A strategy on how to transfer data from the old Elastic to the new Elasticsearch
2. A strategy on how to close the lag between the start of the migration and the end of it

And our two options in closing the lag:

1. Our messaging system is Kafka based. We could have just taken the current offset before the migration started, and after the migration ended, start consuming from that specific offset. This solution requires some manual tweaking of offsets and some other stuff, but will work.
2. Another approach to solving this issue was to start consuming messages from the beginning of the topic in Kafka and make our actions on Elasticsearch idempotent — meaning, if the change was “applied” already, nothing would change in Elastic store.

The requests made by our service against Elastic were already idempotent, so we choose option 2 because it required zero manual work (no need to take specific offsets, and then set them afterward in a new consumer group).

How can we migrate the data?

These were the options we thought of:

1. If our Kafka contained all messages from the beginning of time, we could just play from the start and the end state would be equal. But since we apply retention to out topics, this was not an option.
2. Dump messages to disk and then ingest them to Elastic directly – This solution looked kind of weird. Why store them in disk instead of just writing them directly to Elastic?
3. Transfer messages between old Elastic to new Elastic — This meant, writing some sort of “script” (did anyone say Python? ?) that will connect to the old Elasticsearch cluster, query for items, transform them to the new schema, and index them in the cluster.

We choose the last option. These were the design choices we had in mind:

1. Let’s not try to think about error handling unless we need to. Let’s try to write something super simple, and if errors occur, let’s try to address them. In the end, we did not need to address this issue since no errors occurred during the migration.
2. It’s a one-off operation, so whatever works first / KISS.
3. Metrics — Since the migration processes can take hours to days, we wanted the ability from day 1 to be able to monitor the error count and to track the current progress and copy rate of the script.

We thought long and hard and choose Python as our weapon of choice. The final version of the code is below:

dictor==0.1.2 - to copy and transform our Elasticsearch documentselasticsearch==1.9.0 - to connect to "old" Elasticsearchelasticsearch6==6.4.2 - to connect to the "new" Elasticsearchstatsd==3.3.0 - to report metrics

from elasticsearch import Elasticsearch
from elasticsearch6 import Elasticsearch as Elasticsearch6
import sys
from elasticsearch.helpers import scan
from elasticsearch6.helpers import parallel_bulk
import statsd

ES_SOURCE = Elasticsearch(sys.argv[1])
ES_TARGET = Elasticsearch6(sys.argv[2])
INDEX_SOURCE = sys.argv[3]
INDEX_TARGET = sys.argv[4]
QUERY_MATCH_ALL = {"query": {"match_all": {}}}
SCAN_SIZE = 1000
SCAN_REQUEST_TIMEOUT = '3m'
REQUEST_TIMEOUT = 180
MAX_CHUNK_BYTES = 15 * 1024 * 1024
RAISE_ON_ERROR = False


def transform_item(item, index_target):
    # implement your logic transformation here
    transformed_source_doc = item.get("_source")
    return {"_index": index_target,
            "_type": "_doc",
            "_id": item['_id'],
            "_source": transformed_source_doc}


def transformedStream(es_source, match_query, index_source, index_target, transform_logic_func):
    for item in scan(es_source, query=match_query, index=index_source, size=SCAN_SIZE,
                     timeout=SCAN_REQUEST_TIMEOUT):
        yield transform_logic_func(item, index_target)


def index_source_to_target(es_source, es_target, match_query, index_source, index_target, bulk_size, statsd_client,
                           logger, transform_logic_func):
    ok_count = 0
    fail_count = 0
    count_response = es_source.count(index=index_source, body=match_query)
    count_result = count_response['count']
    statsd_client.gauge(stat='elastic_migration_document_total_count,index={0},type=success'.format(index_target),
                        value=count_result)
    with statsd_client.timer('elastic_migration_time_ms,index={0}'.format(index_target)):
        actions_stream = transformedStream(es_source, match_query, index_source, index_target, transform_logic_func)
        for (ok, item) in parallel_bulk(es_target,
                                        chunk_size=bulk_size,
                                        max_chunk_bytes=MAX_CHUNK_BYTES,
                                        actions=actions_stream,
                                        request_timeout=REQUEST_TIMEOUT,
                                        raise_on_error=RAISE_ON_ERROR):
            if not ok:
                logger.error("got error on index {} which is : {}".format(index_target, item))
                fail_count += 1
                statsd_client.incr('elastic_migration_document_count,index={0},type=failure'.format(index_target),
                                   1)
            else:
                ok_count += 1
                statsd_client.incr('elastic_migration_document_count,index={0},type=success'.format(index_target),
                                   1)

    return ok_count, fail_count


statsd_client = statsd.StatsClient(host='localhost', port=8125)

if __name__ == "__main__":
    index_source_to_target(ES_SOURCE, ES_TARGET, QUERY_MATCH_ALL, INDEX_SOURCE, INDEX_TARGET, BULK_SIZE,
                           statsd_client, transform_item)

Conclusion

Migrating data in a live production system is a complicated task that requires a lot of attention and careful planning. I recommend taking the time to work through the steps listed above and figure out what works best for your needs.

As a rule of thumb, always try to reduce your requirements as much as possible. For example, is a zero downtime migration required? Can you afford data-loss?

Upgrading data stores is usually a marathon and not a sprint, so take a deep breath and try to enjoy the ride.

• The whole process listed above took me around 4 months of work
• All of the Elasticsearch examples that appear in this post have been tested against version 6.8.1

Today, I am going to show you how to implement Elasticsearch in Go.
But of course, before that I am going to give a small introduction to Elasticsearch.
If you have already gained a basic understanding of Elasticsearch, you can skip to the next part.

Elasticsearch

Elasticsearch has been gaining a lot of popularity lately. Searching in a Relational-Database always has issues around scalability and performance.

Elasticsearch is a NoSQL database that has been very successful in tackling those issues. It provides great scalability and performance, and one of the most prominent features is the scoring system that allows a lot of flexibility in the search results. After all, it is not called Elastic-search for no reason!

Installing Elasticsearch

First, you will need to install Elasticsearch on your local machine. You can go to their website and get the installation guide for it. At the time I am writing this article, I am using Elasticsearch with the version number of 7.4.2 .

Elasticsearch has been making a lot of changes in their versions, one of them being the removal of mapping type. So do not expect this to fully work if you are using another version of Elasticsearch.

After finishing your installation, do not forget to run your elasticsearch service, which is mentioned quite clearly on their installation guide (for linux, in short do this ./bin/elasticsearch ).

Make sure your elasticsearch is running by requesting into port 9200 in your local machine.

GET localhost:9200

Hitting it should show something like below.

{
  "name": "204371",
  "cluster_name": "elasticsearch",
  "cluster_uuid": "8Aa0PznuR1msDL9-PYsNQg",
  "version": {
    "number": "7.4.2",
    "build_flavor": "default",
    "build_type": "tar",
    "build_hash": "2f90bbf7b93631e52bafb59b3b049cb44ec25e96",
    "build_date": "2019-10-28T20:40:44.881551Z",
    "build_snapshot": false,
    "lucene_version": "8.2.0",
    "minimum_wire_compatibility_version": "6.8.0",
    "minimum_index_compatibility_version": "6.0.0-beta1"
  },
  "tagline": "You Know, for Search"
}

If it's showing correctly then congratulations! You have successfully run your elasticsearch service in your local machine. Give yourself a clap and take a cup of coffee, since the day is still young.

Making your first index

In Elasticsearch, index is similar to a database. Whereas before, there was table in elasticsearch called type. But since type has been removed in the current version, there are only index now.

Confused now? Don't be. In a nutshell, just think that you only need index then afterwards you just need to insert your data into Elasticsearch.

Now, we are going to make an index named students by doing the query below.
PUT localhost/9200/students

{
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 1
    },
   "mappings": {
       "properties": {
         "name": {
               "type": "text"
         },
         "age": {
               "type": "integer"      
         },
         "average_score": {
               "type": "float"
         }
     }
   }
}

If nothing goes wrong, it should respond back by giving this.

{
    "acknowledged": true,
    "shards_acknowledged": true
}

Your index should be created. Now we will proceed to our next step: playing around with our Elasticsearch index.

Populating your Elasticsearch

First, what we will be doing now is filling in our Elasticsearch index with documents. If you are not familiar with that definition, just know that it is very similar to rows in a database.

In a NoSQL database, it's actually possible for every document to contain different fields that don't match with the schema.

But let's not do that – let's construct our column with a schema that we have defined before. The previous API will allow you to fill the document in your index.

POST localhost:9200/students/doc

{
    "name":"Alice",
    "age":17,
    "average_score":81.1
}

Your Elasticsearch should have one document by now. We will need to insert several more data into our Elasticsearch. And of course, we are not going to insert our student data one by one - that would be quite a hassle!

Elasticsearch has specifically prepared a bulk API in order to send multiple requests at once. Let's use that to insert multiple data at once.

POST /students/_bulk

{ "index":{"_index": "students" } }
{ "name":"john doe","age":18, "average_score":77.7 }
{ "index":{"_index": "students" } }
{ "name":"bob","age":16, "average_score":65.5 }
{ "index":{"_index": "students" } }
{ "name":"mary doe","age":18, "average_score":97.7 }
{ "index":{"_index": "students" } }
{ "name":"eve","age":15, "average_score":98.9 }

Let's query for the data

We have finally populated our Elasticsearch with several more students' data. Now let's do what Elasticsearch is known for: we will try to search our Elasticsearch for the data that we just inserted.

Elasticsearch supports many types of search mechanisms, but for this example we will be using a simple matching query.

Let's start our search by hitting this API:

POST localhost:9200/_search

{
    "query" : {
        "match" : { "name" : "doe" }
    }
}

You will get back your response together with the students' data that matched with your corresponding query. Now you are officially a Search Engineer!

{
    "took": 608,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 2,
            "relation": "eq"
        },
        "max_score": 0.74487394,
        "hits": [
            {
                "_index": "students",
                "_type": "_doc",
                "_id": "rgpef24BTFuh7kXolTpo",
                "_score": 0.74487394,
                "_source": {
                    "name": "john doe",
                    "age": 18,
                    "average_score": 77.7
                }
            },
            {
                "_index": "students",
                "_type": "_doc",
                "_id": "sApef24BTFuh7kXolTpo",
                "_score": 0.74487394,
                "_source": {
                    "name": "mary doe",
                    "age": 18,
                    "average_score": 97.7
                }
            }
        ]
    }
}

Now let's get to Go!

Go in action!

If you have reached this part, you should have grasped the very minimum concepts of using Elasticsearch. Now, we are going to implement Elasticsearch in Go.

A very primitive way of implementing Elasticsearch is that you can keep doing http requests into your Elasticsearch IP. But we are not going to do that.

I found this very helpful library for implementing Elasticsearch in Go. You should install that library before you proceed in your Go modules.

Make your struct

First of all, you will definitely need to make a struct for your Model. In this example, we are going to use the same modeling as in our previous example which in this case is the Student struct.

package main

type Student struct {
    Name         string  `json:"name"`
    Age          int64   `json:"age"`
    AverageScore float64 `json:"average_score"`
}

Making a Client Connection

Now, let's make a function that'll allow us to initialize our ES Client connection.
If you have a running instance of Elasticsearch outside of your localhost, you can simply change the part inside SetURL.

func GetESClient() (*elastic.Client, error) {

    client, err :=  elastic.NewClient(elastic.SetURL("http://localhost:9200"),
        elastic.SetSniff(false),
        elastic.SetHealthcheck(false))

    fmt.Println("ES initialized...")

    return client, err

}

Data Insertion

After that, the first thing we can do is try to insert our data into Elasticsearch via Go. We will be making a model of Student and inserting it into our Elasticsearch client.

package main

import (
    "context"
    "encoding/json"
    "fmt"

    elastic "gopkg.in/olivere/elastic.v7"
)

func main() {

    ctx := context.Background()
    esclient, err := GetESClient()
    if err != nil {
        fmt.Println("Error initializing : ", err)
        panic("Client fail ")
    }

    //creating student object
    newStudent := Student{
        Name:         "Gopher doe",
        Age:          10,
        AverageScore: 99.9,
    }

    dataJSON, err := json.Marshal(newStudent)
    js := string(dataJSON)
    ind, err := esclient.Index().
        Index("students").
        BodyJson(js).
        Do(ctx)

    if err != nil {
        panic(err)
    }

    fmt.Println("[Elastic][InsertProduct]Insertion Successful")

}

Querying our Data

Finally, we can do some searching. The below code might look a bit complex. But rest assured, it will make more sense to you after you go through it carefully. I will be using a basic matching query in the below example.

package main

import (
    "context"
    "encoding/json"
    "fmt"

    elastic "gopkg.in/olivere/elastic.v7"
)

func main() {

    ctx := context.Background()
    esclient, err := GetESClient()
    if err != nil {
        fmt.Println("Error initializing : ", err)
        panic("Client fail ")
    }

    var students []Student

    searchSource := elastic.NewSearchSource()
    searchSource.Query(elastic.NewMatchQuery("name", "Doe"))

    /* this block will basically print out the es query */
    queryStr, err1 := searchSource.Source()
    queryJs, err2 := json.Marshal(queryStr)

    if err1 != nil || err2 != nil {
        fmt.Println("[esclient][GetResponse]err during query marshal=", err1, err2)
    }
    fmt.Println("[esclient]Final ESQuery=\n", string(queryJs))
    /* until this block */

    searchService := esclient.Search().Index("students").SearchSource(searchSource)

    searchResult, err := searchService.Do(ctx)
    if err != nil {
        fmt.Println("[ProductsES][GetPIds]Error=", err)
        return
    }

    for _, hit := range searchResult.Hits.Hits {
        var student Student
        err := json.Unmarshal(hit.Source, &student)
        if err != nil {
            fmt.Println("[Getting Students][Unmarshal] Err=", err)
        }

        students = append(students, student)
    }

    if err != nil {
        fmt.Println("Fetching student fail: ", err)
    } else {
        for _, s := range students {
            fmt.Printf("Student found Name: %s, Age: %d, Score: %f \n", s.Name, s.Age, s.AverageScore)
        }
    }

}

The query should be printed out like this:

ES initialized...
[esclient]Final ESQuery=
 {"query":{"match":{"name":{"query":"Doe"}}}}

And yes that query is what will be posted into the Elasticsearch.

The result of your query should also come out like this if you have followed my example since the very start:

Student found Name: john doe, Age: 18, Score: 77.700000 
Student found Name: mary doe, Age: 18, Score: 97.700000 
Student found Name: Gopher doe, Age: 10, Score: 99.900000

And there you go!

That's the end of my tutorial about how to implement Elasticsearch in Go. I hope I have covered the very basic parts of using Elasticsearch in Go.

To get further info on this topic, you should read about Query DSL and Function Scoring in Elasticsearch, which in my opinion one of the best things about Elasticsearch.

And fret not, the library used in this example also supports a lot of Elasticsearch features, even the Function Scoring query in Elasticsearch.

Thanks for reading through my article! I do hope it will be useful and can help you getting started using Elasticsearch.

Never stop learning; knowledge doubles every fourteen months. ~Anthony J.D'Angelo

The goal is to turn data into information, and information into insight.

―Carly Fiorina

About Kibana

Kibana is a piece of data visualization software that provides a browser-based interface for exploring Elasticsearch data and navigating the Elastic Stack — a collection of open-source products (Elasticsearch, Logstash, Beats, and others).

While Logstash and Bits deliver data to Elasticsearch, Kibana opens the window into the Elastic Stack, allowing you to track the health of your cluster, perform log and time-series analysis, detect anomalies in the data with unsupervised machine learning, discover relationships using graphs and, most importantly, extract insights from the Elasticsearch data with visualizations that can be combined together in a custom interactive dashboard.

Today I’d like to show you how to create a stunning dashboard and a tabular report based on the Elasticsearch data.

Roll up your sleeves and let’s start!

Where to start

The Home page is the place where everything starts.

Here you can decide which actions to take next. The available functionality can be divided into two logical sections:

• Visualizing and exploring the data. Here you can create a new dashboard, visualization or presentation, build a machine learning model, analyze relationships in your data using graphs, and more.
• Managing the Elastic Stack: configure your spaces, analyze logs of an application, configure security settings, etc.

We’ll focus on the process of creating visualizations and adding them to the dashboard.

How to create a dashboard in Kibana

Let me get you a feel for how easy it is to set up a rich dashboard and start reporting.

The first essential step to take is to import your data into Kibana. Multiple options for adding data are at your disposal — you can choose the one that works best for you:

For demonstration purposes, I’ve selected the sample data.

To design your first data visualizations and combine them into the dashboard, open the Visualize page. Here you can create, modify and view the existing visualizations.

What will strike you at once is the abundance of visualization types you can choose from.

After you’ve selected the one you need, choose an index pattern as a source so as to inform Kibana about your index. Let’s choose kibana_sample_data_flights and start creating a horizontal bar chart.

Now you can apply a metric aggregation for the Y-axis and a bucket aggregation for the X-axis. Here is a list of all available aggregations for charts.

Creating a horizontal bar chart in Kibana

Optionally, you can customize the colors of the visualization.

Filtering is another mighty feature of Elasticsearch and Kibana. It provides a way to visualize only a selected subset of documents.

See how you can apply filters to the fields based on logical conditions:

As you see, Kibana provides a straightforward way of filtering the data via the comfy interface. Along with that, you can choose how to filter the data — either by using the Kibana Query Language (a simplified query syntax) or Lucene.

To allow end-users to filter the data interactively, you can add control widgets — special elements of the dashboard which allow filtering the data simply by clicking them.

Another feature I’d like to highlight is the advanced filtering by dates and the ability to set time intervals for refreshing the data in the dashboard.

The good thing is that visualizations are reusable. After creating it, you can save your result and add it to the dashboard any time as well as share with your colleagues given they have access to your Kibana instance.

Saving a visualization in Kibana

After arranging all the visualization elements on a single page, you can export the final dashboard to PNG or PDF format. This is what makes the dashboards portable — it’s easy to share them across departments in no time.

Let’s look at an example of the dashboard you can create:

Interacting with the dashboard in Kibana

To my mind, the principal features which make each dashboard special are interactivity and expressiveness. With it, you can communicate business metrics efficiently.

Personal impression

The visualizations in Kibana ideally perform the tasks they are designed for. What is more, all the visualizations are eye-catching and you can tailor them according to your design ideas. The entire process of creating a dashboard in Kibana is meant to be fast and efficient — and it is so due to the Kibana’s user-friendly and intuitive interface.

On the other hand, I’ve felt that some functionality is missing here.

When working with data, one of the effective exploratory techniques you can apply is slicing and dicing your data before getting to know which aspects of the data to pay attention to. To my mind, the data table widget isn’t the best option — it presents the data in a flat table which doesn’t support a multi-dimensional view of the data. But playing with data should be done interactively and fast.

And this is where a pivot table control comes into play. After searching for available solutions, my choice fell on one open-source plugin called Flexmonster. It handles connecting to the Elasticsearch index and allows creating tabular reports based on the data from its documents. Along with that, integrating with Kibana is smooth — the only thing required to get started is to install a plugin by running one line of code in the command line. You can find more details on GitHub. Before using it, I recommend making sure that your Kibana and Elasticsearch instances are of the same version.

Once you set up a tool, you are ready to use all available features for searching in-depth insights.

Features for analytics and reporting

Flexmonster Pivot provides fast access to the most essential reporting functionality. Its toolbar allows connecting to the data source, loading previously saved reports, exporting reports to PDF, Excel, HTML, CSV, and images. Besides, I’ve managed to quickly switch between two different modes — the grid and the charts. Cells formatting options include conditional and number formatting. The field list deserves particular attention — here you can select hierarchies to rows, columns, measures, and report filters. There is also the search input field which is helpful if the index has a long list of fields.

One of the features I’d like to highlight is the ability to drag and drop the hierarchies right on the grid. Thereby, you can change the slice completely via the UI.

Another one is the drill-through feature — it helps to know which records stand behind the aggregated values.

Working with a pivot table

Let me show you how to create a report based on the Elasticsearch data:

While testing the tool, I’ve managed to aggregate and filter the data, sort the values on the grid and save the results to continue working with the report later. Plus, exporting works well — it’s easy to share the reports with teammates.

Bringing it all together

Today I’ve covered the benefits Kibana provides for visualization of Elasticsearch data. You’ve been able to make sure how dashboards can empower the analysis process.

To my mind, a pivot table is a good tool which enables you to benefit from exploring data before teasing out the answers to complex questions.

Flexmonster nicely complements the available functionality of Kibana - the reports you are creating with it are insightful, customizable and can be easily shared across departments.

Working together, both tools have all the potential to boost your storytelling.

I encourage you to give such a combination a try.

What’s next?

• Reporting with Kibana
• Creating a visualization in Kibana
• Pivot Table for Elasticsearch
• How to add a Pivot Table to Kibana

Logging is an essential component within any application. Logs enable you to analyze and sneak a peak into what’s happening within your application code like a story. Software developers spend a large part of their day to day lives monitoring, troubleshooting and debugging applications, which can sometimes be a nightmare. Logging allows software developers to make this hectic process much easier and smoother.

If you have containerized your application with a container platform like Docker, you may be familiar with docker logs which allows you to see the logs created within your application running inside your docker container. Why then think of Elastic Stack to analyze your logs? Well, there are mainly two burning problems here:

• Imagine you have tens, hundreds, or even thousands of containers generating logs — SSH-ing in to all those servers and extracting logs won’t work well.
• Also containers are immutable and ephemeral, which means they have a shorter life span. So once your containers are gone and replaced with new containers, all of your application logs related to old containers are gone.

So the ultimate solution for this is to create a centralized logging component for collecting all of your container logs into a single place. This is where Elastic Stacks comes in.

Elastic Stack mainly consists of four major components:

• Beats is the new member which made the ELK Stack known as Elastic Stack. Beats are light weight log data shippers which can push logs to the ELK Stack. For this post I will be using Filebeats, a member of the Beats family, which offers a lightweight way to collect and forward and centralize logs and files.
• Logstash is a component which aggregates, modifies, and transfers logs from multiple input locations into Elasticsearch.
• Elasticsearch is a distributed, JSON-based search and analytics engine that stores and indexes data (log entries in this case) in a scalable and manageable way.
• Kibana is an enriched UI to analyze and easily access data in Elasticsearch.

In this post, we will look into how to use the above mentioned components and implement a centralized log analyzer to collect and extract logs from Docker containers.

For the purposes of this article, I have used two t2.small AWS EC2 instances, running Ubuntu 18.04 installed with Docker and Docker compose. Instance 1 is running a tomcat webapp and the instance 2 is running ELK stack (Elasticsearch, Logstash, Kibana).

In Linux by default docker logs can be found in this location:
/var/lib/docker/containers//<container-id&gt;-json.log

All docker logs will be collected via Filebeat running inside the host machine as a container. Filebeat will be installed on each docker host machine (we will be using a custom Filebeat docker file and systemd unit for this which will be explained in the Configuring Filebeat section.)

Our tomcat webapp will write logs to the above location by using the default docker logging driver. Filebeat will then extract logs from that location and push them towards Logstash.

Another important thing to note is that other than application generated logs, we also need metadata associated with the containers, such as container name, image, tags, host etc… This will allow us to specifically identify the exact host and container the logs are generating. These data can also be sent easily by Filebeat along with the application log entries.

High Level Architecture — Instance 1 [Left] | Instance 2 [Right]

By doing this kind of implementation the running containers don’t need to worry about the logging driver, how logs are collected and pushed. Filebeat will take care of those. This is often known as single responsibility principle.

Configuring Filebeat

For this section the filebeat.yml and Dockerfile were obtained from Bruno COSTE’s sample-filebeat-docker-logging github repo. Many thanks to his awesome work.

But since I have done several changes to filebeat.yml according to requirements of this article, I have hosted those with filebeat.service (systemd file) separately on my own repo. You can access the repo here.

As the initial step, you need to update your filebeat.yml file which contains the Filebeat configurations. Given below is a sample filebeat.yml file you can use. Note the line 21, the output.logstash field and the hosts field. I have configured it to the IP address of the server I’m running my ELK stack, but you can modify it if you are running Logstash on a separate server. By default Logstash is listening to Filebeat on port 5044.

To get to know more about Filebeat Docker configuration parameters, look here.

After that you can create your own Filebeat Docker image by using the following Dockerfile.

Once the image is built, you can push it in to your docker repository. Now since you have the capability to run Filebeat as a docker container, it’s just a matter of running the Filebeat container on your host instances running containers. Here is the docker run command.

docker run -v '/var/lib/docker/containers:/usr/share/dockerlogs/data:ro' -v '/var/run/docker.sock:/var/run/docker.sock' --name filebeat ${YOUR_FILEBEAT_DOCKER_IMAGE}:latest

In the above Docker command, note the two bind mount parameters: /var/lib/docker/containers is the path where docker logs exist within the host machine, and it has been bound to /usr/share/dockerlogs/data path within Filebeat container with read only access. In the second bind mount argument, /var/run/docker.sock is bound into the Filebeat container’s Docker daemon. It is the unix socket the Docker daemon listens on by default and it can be used to communicate with the daemon from within a container. This allows our Filebeat container to obtain Docker metadata and enrich the container log entries along with the metadata and push it to ELK stack.

If you want to automate this process, I have written a Systemd Unit file for managing Filebeat as a service.

Configuring the ELK Stack

For this I will be using my second EC2 instance, where I run the ELK stack. You can do this by simply installing Docker compose and checking out this awesome deviantony/docker-elk repo and just running docker-compose up -d

Note that all your firewall rules allow inbound traffic into the Logstash, Elasticsearch and Kibana.

Before running the ELK stack you need to make sure your logstash.conf file is properly configured to listen to incoming beats logs on port 5044 and the logs are being properly added onto the elasticsearch host. Also you need to make sure to add an index parameter on to your Elasticsearch to identify the logs generated by Filbeat uniquely.

In your docker-elk repo you can find your logstash.conf file by following docker-elk/logstash/pipeline pathname. This is the configuration file for setting up Logstash configurations. You need to update it as follows:

Once you do it, you can access your Kibana dashboard on port 5601 by default as defined on the docker-compose.yml file on deviantony/docker-elk repo.

Kibana Dashboard

Under the management tab, you can create an index pattern for Filebeat logs. This has to be done before you can view the logs on Kibana dashboard.

Filebeat Index Patten Configuration on Kibana Dashboard

If your containers are pushing logs properly into Elasticsearch via Logstash, and you have successfully created the index pattern, you can go to the Discover tab on the Kibana dashboard and view your Docker container application logs along with Docker metadata under the filebeat* index pattern.

Discover Docker container application logs along with the Docker host metadata in Kibana Dashboard

References

1. https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-getting-started.html
2. https://medium.com/@bcoste/powerful-logging-with-docker-filebeat-and-elasticsearch-8ad021aecd87
3. https://www.elastic.co/guide/en/logstash/current/configuration.html
4. https://medium.com/lucjuggery/about-var-run-docker-sock-3bfd276e12fd

_The [GitXplore](https://appbaseio-apps.github.io/gitxplore-app/" rel="noopener" target="blank" title=") app

Elasticsearch is one of the most popular full-text search engines which allows you to search huge volumes of data quickly, while React is arguably the best library for building user interfaces. During the past few months I’ve been co-authoring an open-source library, ReactiveSearch, which provides React components for Elasticsearch and simplifies the process of building a search User Interface (UI).

This is the app which I’ll be building in this story:

_Check out the app on [CodeSandbox](https://codesandbox.io/s/github/appbaseio-apps/gitxplore-app/tree/master/" rel="noopener" target="blank" title=")

A brief idea of Elasticsearch

Elasticsearch is a NoSQL database which can search through large amounts of data in a short time. It performs a full-text search on the data which is stored in the form of documents (like objects) by examining all the words in every document.

Here’s what the Elasticsearch docs say:

Elasticsearch is a highly scalable open-source full-text search and analytics engine. It allows you to store, search, and analyze big volumes of data quickly and in near real time.

Even if you’ve never used Elasticsearch before you should be able to follow along with this story and build your very own Elasticsearch powered search using React and ReactiveSearch. ?

What is ReactiveSearch?

ReactiveSearch is a React UI components library for Elasticsearch. In order to search data in Elasticsearch, you need to write queries. Then you will need to format and render the JSON data in your UI. ReactiveSearch simplifies the entire process since you don’t need to worry about writing these queries. This makes it easier to focus on creating the UI.

Here is an example that generates a search-box UI with category specific suggestions:

<CategorySearch
  componentId="repo"
  dataField={["name", "name.raw"]}
  categoryField="language.raw"
/>

Component rendered from the above code

This would likely have taken us 100+ lines without the library, and knowledge of Elasticsearch Query DSL to construct the query.

In this post, I’ll use different components from the library to build the final UI.

You should try out the final app before we deep-dive. Here’s the CodeSandbox link for the same.

Setting things up

Before we start building the UI, we’ll need the dataset containing GitHub repositories in Elasticsearch. ReactiveSearch works with any Elasticsearch index and you can easily use it with your own dataset.

For brevity, you can use my dataset or clone it for yourself by following this link and clicking on Clone this App button. This will let you make a copy of the dataset as your own app.

_The GitHub repo [dataset](https://opensource.appbase.io/dejavu/live/#?input_state=XQAAAAJDAQAAAAAAAAA9iIqnY-B2BnTZGEQz6wkFsoFSyhi0TotY1ZI3dCbzpZ5wZmCa4HoWjWiBHcRO1KpPWzrR3-ungbYF_FBD7IY3vlhuTW9dQQFtt3qksr-wGqyFf_qxW2Z3widjMRY5xGpv9lCIh4b5Dyi-O2wVMmUzKADc-0pG1tyzQ558Y_SoViZ27V2qq-px_fIGV-GVRTcrO-LdiYhDhtFK4tYVTak07UxRRvGaqeK3GI2sU7O67YnSdDZNv8_5pnc3SPxlPV9t9YdkGW3YkckG3LAVp03TbrSWI7GdN0fMZCgwqWv0FP1iNWHQrUW2v8-B___Y4BHg" rel="noopener" target="blank" title=")

After you enter an app name, the cloning process should start importing the 26K+ repos to your account.

All the repos are structured in the following format:

{
  "name": "freeCodeCamp",
  "owner": "freeCodeCamp",
  "fullname": "freeCodeCamp~freeCodeCamp",
  "description": "The https://freeCodeCamp.org open source codebase and curriculum. Learn to code and help nonprofits.",
  "avatar": "https://avatars0.githubusercontent.com/u/9892522?v=4",
  "url": "https://github.com/freeCodeCamp/freeCodeCamp",
  "pushed": "2017-12-24T05:44:03Z",
  "created": "2014-12-24T17:49:19Z",
  "size": 31474,
  "stars": 291526,
  "forks": 13211,
  "topics": [
    "careers",
    "certification",
    "community",
    "curriculum",
    "d3",
    "education",
    "javascript",
    "learn-to-code",
    "math",
    "nodejs",
    "nonprofits",
    "programming",
    "react",
    "teachers"
  ],
  "language": "JavaScript",
  "watchers": 8462
}

• We will use create-react-app to set up the project. You can install create-react-app by running the following command in your terminal:

npm install -g create-react-app

• After it’s installed, you can create a new project by running:

create-react-app gitxplore

• After the project is set up you can change into the project directory and install ReactiveSearch dependency:

cd gitxplore
npm install @appbaseio/reactivesearch

• You may also add fontawesome CDN, which we’ll be using for some icons, by inserting the following lines in /public/index.html before the </body> tag ends:

<script defer         src="https://use.fontawesome.com/releases/v5.0.2/js/all.js"></script>

Diving into the code

I’ll follow a simple directory structure for the app. Here are the important files:

src
├── App.css               // App styles
├── App.js                // App container
├── components
│   ├── Header.js         // Header component
│   ├── Results.js        // Results component
│   ├── SearchFilters.js  // Filters component
│   └── Topic.js          // rendered by Results
├── index.css             // styles
├── index.js              // ReactDOM render
└── theme.js              // colors and fonts
public
└── index.html

Here’s the link to final repo if you wish to reference anything at any point.

1. Adding styles

I’ve written responsive styles for the app which you can copy into your app. Just fire up your favorite text editor and copy the styles for /src/index.css from here and /src/App.css from here respectively.

Now, create a file /src/theme.js where we’ll add the colors and fonts for our app:

const theme = {
    typography: {
        fontFamily: 'Raleway, Helvetica, sans-serif',
    },
    colors: {
        primaryColor: '#008000',
        titleColor: 'white'
    },
    secondaryColor: 'mediumseagreen',
};

export default theme;

2. Adding the first ReactiveSearch component

All the ReactiveSearch components are wrapped around a container component ReactiveBase which provides data from Elasticsearch to the children ReactiveSearch components.

We’ll use this in /src/App.js:

import React, { Component } from 'react';
import { ReactiveBase } from '@appbaseio/reactivesearch';
import theme from './theme';
import './App.css';
class App extends Component {
  render() {
    return (
      <section className="container">
        <ReactiveBase
          app="gitxplore-app"
          credentials="4oaS4Srzi:f6966181-1eb4-443c-8e0e-b7f38e7bc316"
          type="gitxplore-latest"
          theme={theme}
        >
          <nav className="navbar">
            <div className="title">GitXplore</div>
          </nav>
        </ReactiveBase>
      </section>
    );
  }
}
export default App;

For the app and credentials prop you may use the ones I’ve provided here as it is. If you cloned the dataset in your own app earlier you can get them from the app’s credentials page. If you’re already familiar with Elasticsearch you may instead pass a url prop referring to your own Elasticsearch cluster URL.

_Getting app’s credentials from appbase.io [dashboard](https://dashboard.appbase.io/credentials" rel="noopener" target="blank" title="). Just copy the Read-only API key

Alternatively, you can also copy your app’s credentials from the apps dashboard. Hover over your app’s card and click on Copy Read Credentials.

_Alternative to above link: Copy the read credentials from [apps dashboard](https://dashboard.appbase.io/apps" rel="noopener" target="blank" title=")

After adding this you would see a basic layout like this:

After adding the first ReactiveSearch component

3. Adding a DataSearch

DataSearch component

Next, I’ll be adding a DataSearch component to search through repositories. It creates a search UI component and lets us search across one or more fields easily. The updated render function in /src/App.js would look like this:

// importing DataSearch here
import { ReactiveBase, DataSearch } from '@appbaseio/reactivesearch';
...
<ReactiveBase ... >
// Adding the DataSearch here
    <div className="flex row-reverse app-container">
        <div className="results-container">
            <DataSearch
                componentId="repo"
                filterLabel="Search"
                dataField={['name', 'description', 'name.raw', 'fullname', 'owner', 'topics']}
                placeholder="Search Repos"
                autosuggest={false}
                iconPosition="left"
                URLParams
                className="data-search-container results-container"
                innerClass={{
                    input: 'search-input',
                }}
            />
        </div>
    </div>
</ReactiveBase>
...

The DataSearch component goes inside the ReactiveBase component and receives all the necessary data from it so we don’t have to write Elasticsearch queries ourselves. The surrounding divs add some className properties for styling. These just add a layout to the app. You can go through all the styles at /src/App.css which we created earlier. You might have noticed that we have passed some props to the DataSearch component.

Here’s how they work:

• componentId: a unique string identifier which we’ll use later to connect two different ReactiveSearch components.
• filterLabel: a string value which will show up in the filters menu later.
• dataField: an array of strings containing Elasticsearch fields on which search has to performed on. You can check the dataset and see that these fields also matches the column name. All fields specified here matches the structure of data, for example name refers to the name of repo, description refers to its description, but there is a field with a .raw added here, name.raw which is a multi-field of the name field. Elasticsearch can index the same data in different ways for different purposes, which we can use to get better search results.
• placeholder: sets the placeholder value in the input box.
• autosuggest: setting a false value for the prop causes the results to update immediately in the results.
• iconPosition: sets the position of the ? icon.
• URLParams: is a boolean which tells the component to save the search term in the browser’s URL so we can share a URL to a specific search query. For example, check this link to see all results related to “react”.
• className: adds a class for styling using CSS.
• innerClass: adds a class to different sections of a component for styling using CSS. Here, I’ve added a class to the input box for styling. A detailed description can be found in the docs.

With this, our app should get a working search bar:

Adding DataSearch component

4. Adding the Results view

Next, we’ll be adding the Results component at /src/components/Results.js and importing it in /src/App.js.

Here’s how you can write the Results component:

import React from 'react';
import { SelectedFilters, ReactiveList } from '@appbaseio/reactivesearch';
const onResultStats = (results, time) => (
  <div className="flex justify-end">
    {results} results found in {time}ms
  </div>
);
const onData = (data) => (
  <div className="result-item" key={data.fullname}>
    {data.owner}/{data.name}
  </div>
);
const Results = () => (
  <div className="result-list">
    <SelectedFilters className="m1" />
    <ReactiveList
      componentId="results"
      dataField="name"
      onData={onData}
      onResultStats={onResultStats}
      react={{
        and: ['repo'],
      }}
      pagination
      innerClass={{
        list: 'result-list-container',
        pagination: 'result-list-pagination',
        resultsInfo: 'result-list-info',
        poweredBy: 'powered-by',
      }}
      size={6}
    />
  </div>
);
export default Results;

I’ve imported two new components from ReactiveSearch, SelectedFilters and ReactiveList. SelectedFilters will render the filters for our ReactiveSearch components at one place:

SelectedFilters renders removable filters

ReactiveList renders the search results. Here’s how its props work:

• dataField: orders the results using name field here.
• onData: accepts a function which returns a JSX. The function is passed each result individually. Here we’re generating a basic UI which we’ll modify later.
• onResultStats: similar to onData but for the result stats. The function is passed the number of results found and time taken.
• react: the [react](https://opensource.appbase.io/reactive-manual/advanced/react.html) prop tells the ReactiveList to listen to changes made byCategorySearch component, we’ve provided the componentId of the CategorySearch component here called repo. Later we’ll add more components here.
• pagination: a boolean which tells the ReactiveList to split the results into pages, each page containing the number of results specified in the size prop.

Now we can import and use the Results component in /src/App.js. Just add it inside the div with results-container class.

...
import Results from './components/Results';
...
render() {
  return(
    ...
    <div className="results-container">
      <DataSearch ... />
      <Results />
    </div>
    ...
  )
}

With this component, a basic version of our search UI should start coming together:

Adding the Results component

5. Adding a Header component

Lets create a Header component at /src/components/Header.js which we’ll use to render more search filters.

Here’s how to create a simple Header component:

import React, { Component } from 'react';

import SearchFilters from './SearchFilters';

class Header extends Component {
    constructor(props) {
        super(props);
        this.state = {
            visible: false,
        };
    }

    toggleVisibility = () => {
        const visible = !this.state.visible;
        this.setState({
            visible,
        });
    }

    render() {
        return (
            <nav className={`navbar ${this.state.visible ? 'active' : ''}`}>
                <div className="title">GitXplore</div>
                <div className="btn toggle-btn" onClick={this.toggleVisibility}>Toggle Filters</div>
                <SearchFilters {...this.props} visible={this.state.visible} />
            </nav>
        );
    }
}

export default Header;

I’ve moved the navigation code in <nav>..</nav> from /src/App.js here. The Header component has a method which toggles visible in the state. We’re using this to add a class which would make it take up the entire screen size on mobile layout. I’ve also added a toggle button which calls the toggleVisibility method.

It also renders another component called SearchFilters and passes all the props from the parent App component. Let’s create this component to see things in action.

Create a new file /src/components/SearchFilters.js:

import React from 'react';
const SearchFilters = () => (
    <div>
        Search filters go here!
    </div>
);
export default SearchFilters;

Next, I’ll update the App component to use the Header component that we created just now.

6. Updating App component and handling topics in state

We’ll add a state variable in App component called currentTopics which would be an array of currently selected topics in the app.

We’ll then use the currentTopics and pass them to the Header and Results components:

import React, { Component } from 'react';
import { ReactiveBase, DataSearch } from '@appbaseio/reactivesearch';

import Header from './components/Header';
import Results from './components/Results';

import theme from './theme';
import './App.css';

class App extends Component {
    constructor(props) {
        super(props);
        this.state = {
            currentTopics: [],
        };
    }

    setTopics = (currentTopics) => {
        this.setState({
            currentTopics: currentTopics || [],
        });
    }

    toggleTopic = (topic) => {
        const { currentTopics } = this.state;
        const nextState = currentTopics.includes(topic)
            ? currentTopics.filter(item => item !== topic)
            : currentTopics.concat(topic);
        this.setState({
            currentTopics: nextState,
        });
    }

    render() {
        return (
            <section className="container">
                <ReactiveBase
                    app="gitxplore-app"
                    credentials="4oaS4Srzi:f6966181-1eb4-443c-8e0e-b7f38e7bc316"
                    type="gitxplore-latest"
                    theme={theme}
                >
                    <div className="flex row-reverse app-container">
                        <Header currentTopics={this.state.currentTopics} setTopics={this.setTopics} />
                        <div className="results-container">
                            <DataSearch
                                componentId="repo"
                                filterLabel="Search"
                                dataField={['name', 'description', 'name.raw', 'fullname', 'owner', 'topics']}
                                placeholder="Search Repos"
                                iconPosition="left"
                                autosuggest={false}
                                URLParams
                                className="data-search-container results-container"
                                innerClass={{
                                    input: 'search-input',
                                }}
                            />
                            <Results currentTopics={this.state.currentTopics} toggleTopic={this.toggleTopic} />
                        </div>
                    </div>
                </ReactiveBase>
            </section>
        );
    }
}

export default App;

The setTopics method will set whichever topics are passed to it, which we’ll pass to the Header component. The toggleTopic method will remove a topic from the state in currentTopics if it’s already present and add the topic if it is not present.

We’ll pass the toggleTopic method to the Results component:

Its starting to come together, cheers!

7. Adding more filters

Lets add more filters to the UI in /src/components/SearchFilters.js. I’ll be using three new components from ReactiveSearch here, MultiDropdownList, SingleDropdownRange and RangeSlider. The components are used in a similar fashion as we used the DataSearch component earlier.

Here’s the code:

import React from 'react';
import PropTypes from 'prop-types';
import {
    MultiDropdownList,
    SingleDropdownRange,
    RangeSlider,
} from '@appbaseio/reactivesearch';

const SearchFilters = ({ currentTopics, setTopics, visible }) => (
    <div className={`flex column filters-container ${!visible ? 'hidden' : ''}`}>
        <div className="child m10">
            <MultiDropdownList
                componentId="language"
                dataField="language.raw"
                placeholder="Select languages"
                title="Language"
                filterLabel="Language"
            />
        </div>
        <div className="child m10">
            <MultiDropdownList
                componentId="topics"
                dataField="topics.raw"
                placeholder="Select topics"
                title="Repo Topics"
                filterLabel="Topics"
                size={1000}
                queryFormat="and"
                defaultSelected={currentTopics}
                onValueChange={setTopics}
            />
        </div>
        <div className="child m10">
            <SingleDropdownRange
                componentId="pushed"
                dataField="pushed"
                placeholder="Repo last active"
                title="Last Active"
                filterLabel="Last Active"
                data={[
                    { start: 'now-1M', end: 'now', label: 'Last 30 days' },
                    { start: 'now-6M', end: 'now', label: 'Last 6 months' },
                    { start: 'now-1y', end: 'now', label: 'Last year' },
                ]}
            />
        </div>
        <div className="child m10">
            <SingleDropdownRange
                componentId="created"
                dataField="created"
                placeholder="Repo created"
                title="Created"
                filterLabel="Created"
                data={[
                    {
                        start: '2017-01-01T00:00:00Z',
                        end: '2017-12-31T23:59:59Z',
                        label: '2017',
                    },
                    {
                        start: '2016-01-01T00:00:00Z',
                        end: '2016-12-31T23:59:59Z',
                        label: '2016',
                    },
                    {
                        start: '2015-01-01T00:00:00Z',
                        end: '2015-12-31T23:59:59Z',
                        label: '2015',
                    },
                    {
                        start: '2014-01-01T00:00:00Z',
                        end: '2014-12-31T23:59:59Z',
                        label: '2014',
                    },
                    {
                        start: '2013-01-01T00:00:00Z',
                        end: '2013-12-31T23:59:59Z',
                        label: '2013',
                    },
                    {
                        start: '2012-01-01T00:00:00Z',
                        end: '2012-12-31T23:59:59Z',
                        label: '2012',
                    },
                    {
                        start: '2011-01-01T00:00:00Z',
                        end: '2011-12-31T23:59:59Z',
                        label: '2011',
                    },
                    {
                        start: '2010-01-01T00:00:00Z',
                        end: '2010-12-31T23:59:59Z',
                        label: '2010',
                    },
                    {
                        start: '2009-01-01T00:00:00Z',
                        end: '2009-12-31T23:59:59Z',
                        label: '2009',
                    },
                    {
                        start: '2008-01-01T00:00:00Z',
                        end: '2008-12-31T23:59:59Z',
                        label: '2008',
                    },
                    {
                        start: '2007-01-01T00:00:00Z',
                        end: '2007-12-31T23:59:59Z',
                        label: '2007',
                    },
                ]}
            />
        </div>
        <div className="child m10">
            <RangeSlider
                componentId="stars"
                title="Repo Stars"
                dataField="stars"
                range={{ start: 0, end: 300000 }}
                showHistogram={false}
                rangeLabels={{
                    start: '0 Stars',
                    end: '300K Stars',
                }}
                innerClass={{
                    label: 'range-label',
                }}
            />
        </div>
        <div className="child m10">
            <RangeSlider
                componentId="forks"
                title="Repo Forks"
                dataField="forks"
                range={{ start: 0, end: 180500 }}
                showHistogram={false}
                rangeLabels={{
                    start: '0 Forks',
                    end: '180K Forks',
                }}
                innerClass={{
                    label: 'range-label',
                }}
            />
        </div>
    </div>
);

SearchFilters.propTypes = {
    currentTopics: PropTypes.arrayOf(PropTypes.string),
    setTopics: PropTypes.func,
    visible: PropTypes.bool,
};

export default SearchFilters;

The SearchFilters component we’ve created above takes in three props from the Header component, currentTopics, setTopics and visible. The visible prop is just used to add a className for styling.

The first component we’ve used here is a [MultiDropdownList](https://opensource.appbase.io/reactive-manual/list-components/multidropdownlist.html) which renders a dropdown component to select multiple options. The first MultiDropdownList has a dataField of language.raw. It’ll populate itself with all the languages available in the repositories dataset.

_The language [MultiDropdownList](https://opensource.appbase.io/reactive-manual/list-components/multidropdownlist.html" rel="noopener" target="blank" title=")

We’ve used another MultiDropdownList to render a list of topics:

<MultiDropdownList
    componentId="topics"
    dataField="topics.raw"
    placeholder="Select languages"
    title="Repo Topics"
    filterLabel="Topics"
    size={1000}
    queryFormat="and"
    defaultSelected={currentTopics}
    onValueChange={setTopics}
/>

Here’s how the props work here:

• componentId: similar to the previous ReactiveSearch components, this is a unique identifier which we’ll later associate in the Results component that we created to get search results.
• dataField: maps the component to the topics.raw field in Elasticsearch.
• placeholder: sets the placeholder value when nothing is selected.
• title: adds a title for the component in the UI.
• filterLabel: sets the label of the components in the removable filters (the SelectedFilters which we used in the Results component).
• size: tells the component to render a maximum of 1000 items in the list.
• queryFormat: when set to 'and' as we’ve used here, it gives results which matches all the selected tags (exactly like intersection).
• defaultSelected: sets the selected items in the component. Here we’re passing currentTopics which we’ve stored in the state at /src/App.js.
• onValueChange: is a function that will be called by the component when we make a change in its value. Here we call the setTopics function which we received in the props. Therefore, whenever we select or deselect a value in the component it would update the currentTopics in the state of main App component.

The topics MultiDropdownList component

The next ReactiveSearch component we’ve used here is a [SingleDropdownRange](https://opensource.appbase.io/reactive-manual/range-components/singledropdownrange.html). It uses a new prop called [data](https://opensource.appbase.io/reactive-manual/range-components/singledropdownrange.html#props).

Here’s how it works:

<SingleDropdownRange
    ...
    data={[
        { start: 'now-1M', end: 'now', label: 'Last 30 days' },
        { start: 'now-6M', end: 'now', label: 'Last 6 months' },
        { start: 'now-1y', end: 'now', label: 'Last year' },
    ]}
/>

The data prop accepts an array of objects with start and end values and shows the specified label in the dropdown. It’s mapped to the pushed field in the dataset which is a date type in Elasticsearch. One cool way to specify date range in Elasticsearch is using the now keyword. now refers to the current time, now-1M refers to one month before, now-6M to six month before and now-1y to a year before now.

_The pushed [SingleDropdownRange](https://opensource.appbase.io/reactive-manual/range-components/singledropdownrange.html" rel="noopener" target="blank" title=") component

I’ve used another SingleDropdownRange component for the created field in the dataset.

Here I’ve specified year ranges in datetime for different years:

<SingleDropdownRange
    ...
    data={[
        {
            start: '2017-01-01T00:00:00Z',
            end: '2017-12-31T23:59:59Z',
            label: '2017',
        },
        {
            start: '2016-01-01T00:00:00Z',
            end: '2016-12-31T23:59:59Z',
            label: '2016',
        },
       ...
    ]}
/>

SingleDropdownRange component for the created field

The third component I’ve used is a [RangeSlider](https://opensource.appbase.io/reactive-manual/range-components/rangeslider.html) which renders a slider UI. I’ve used to RangeSlider components, one for the stars field and the other for forks.

Two main props that this component introduces are range and rangeLabels:

<RangeSlider
    ...
    showHistogram={false}
    range={{ start: 0, end: 300000 }}
    rangeLabels={{
        start: '0 Stars',
        end: '300K Stars',
    }}
/>

• range: prop specifies a range for the data with a start and end value.
• rangeLabels: prop takes the labels to show below the slider.
• showHistogram: is a boolean prop which shows a histogram with the distribution of data. Here I’ve set it to false since it’s not needed.

_[RangeSlider](https://opensource.appbase.io/reactive-manual/range-components/rangeslider.html" rel="noopener" target="blank" title=") components for the stars and forks fields

Now we just need to connect these filters to the Results component. We just have to update one line in the ReactiveList rendered by the Results component to include the componentIds of these components.

Update the react prop in the ReactiveList that we rendered in the Results component:

const Results = () => (
  <div className="result-list">
    <SelectedFilters className="m1" />
    <ReactiveList
      ... // updating the react prop here
      react={{
        and: ['language', 'topics', 'pushed', 'created', 'stars', 'forks', 'repo'],
      }}
    />
  </div>
);

That should make your results update for all the filters ?

After connecting the filters in the ReactiveList component

8. Updating the results view

Up until now, we’ve been seeing only a basic version of the results. As the final piece of this app, lets add some flair to the results ✌️

We’ll be using another component inside our Results components to render different topics.

Topics component to render these little guys

Here’s how you can create your own at /src/components/Topic. Feel free to add your own taste ?

import React, { Component } from 'react';
import PropTypes from 'prop-types';

class Topic extends Component {
    handleClick = () => {
        this.props.toggleTopic(this.props.children);
    }
    render() {
        return (
            <div className={`topic ${this.props.active ? 'active' : ''}`} onClick={this.handleClick}>
                #{this.props.children}
            </div>
        );
    }
}

Topic.propTypes = {
    children: PropTypes.string,
    active: PropTypes.bool,
    toggleTopic: PropTypes.func,
};

export default Topic;

This component renders its children and adds a click handler to toggle the topics which updates the currentTopics inside the main App component’s state.

Next, we just need to update our Results component at /src/components/Results.js:

import React from 'react';
import { SelectedFilters, ReactiveList } from '@appbaseio/reactivesearch';
import PropTypes from 'prop-types';

import Topic from './Topic';

const onResultStats = (results, time) => (
    <div className="flex justify-end">
        {results} results found in {time}ms
    </div>
);

const onData = (data, currentTopics, toggleTopic) => (
    <div className="result-item" key={data.fullname}>
        <div className="flex justify-center align-center result-card-header">
            <img className="avatar" src={data.avatar} alt="User avatar" />
            <a className="link" href={data.url} target="_blank" rel="noopener noreferrer">
                <div className="flex wrap">
                    <div>{data.owner}/</div>
                    <div>{data.name}</div>
                </div>
            </a>
        </div>
        <div className="m10-0">{data.description}</div>
        <div className="flex wrap justify-center">
            {
                data.topics.slice(0, 7)
                    .map(item => (
                        <Topic
                            key={item}
                            active={currentTopics.includes(item)}
                            toggleTopic={toggleTopic}
                        >
                            {item}
                        </Topic>
                    ))
            }
        </div>
        <div className="flex">
            <div><div className="btn card-btn"><i className="card-icon fas fa-star" />{data.stars}</div></div>
            <div><div className="btn card-btn"><i className="card-icon fas fa-code-branch" />{data.forks}</div></div>
            <div><div className="btn card-btn"><i className="card-icon fas fa-eye" />{data.watchers}</div></div>
        </div>
    </div>
);

const Results = ({ toggleTopic, currentTopics }) => (
    <div className="result-list">
        <SelectedFilters className="m1" />
        <ReactiveList
            componentId="results"
            dataField="name"
            onData={data => onData(data, currentTopics, toggleTopic)}
            onResultStats={onResultStats}
            react={{
                and: ['language', 'topics', 'pushed', 'created', 'stars', 'forks', 'repo'],
            }}
            pagination
            innerClass={{
                list: 'result-list-container',
                pagination: 'result-list-pagination',
                resultsInfo: 'result-list-info',
                poweredBy: 'powered-by',
            }}
            size={6}
            sortOptions={[
                {
                    label: 'Best Match',
                    dataField: '_score',
                    sortBy: 'desc',
                },
                {
                    label: 'Most Stars',
                    dataField: 'stars',
                    sortBy: 'desc',
                },
                {
                    label: 'Fewest Stars',
                    dataField: 'stars',
                    sortBy: 'asc',
                },
                {
                    label: 'Most Forks',
                    dataField: 'forks',
                    sortBy: 'desc',
                },
                {
                    label: 'Fewest Forks',
                    dataField: 'forks',
                    sortBy: 'asc',
                },
                {
                    label: 'A to Z',
                    dataField: 'owner.raw',
                    sortBy: 'asc',
                },
                {
                    label: 'Z to A',
                    dataField: 'owner.raw',
                    sortBy: 'desc',
                },
                {
                    label: 'Recently Updated',
                    dataField: 'pushed',
                    sortBy: 'desc',
                },
                {
                    label: 'Least Recently Updated',
                    dataField: 'pushed',
                    sortBy: 'asc',
                },
            ]}
        />
    </div>
);

Results.propTypes = {
    toggleTopic: PropTypes.func,
    currentTopics: PropTypes.arrayOf(PropTypes.string),
};

export default Results;

I’ve updated the onData function to render more detailed results. You’ll also notice a new sortOptions prop in the ReactiveList. This prop accepts an array of objects which renders a dropdown menu to select how you wish to sort the results. Each object contains a label to display as the list item, a dataField to sort the results on and a sortBy key which can either be asc (ascending) or desc (descending).

That’s it, your very own GitHub repository explorer should be live!

_GitXplore [final app preview](https://appbaseio-apps.github.io/gitxplore-app/" rel="noopener" target="blank" title=")

Useful links

1. GitXplore app demo, CodeSandbox and source code
2. ReactiveSearch GitHub repo
3. ReactiveSearch docs

Hope you enjoyed this story. If you have any thoughts or suggestions, please let me know and do share your version of the app in comments!

You may follow me on twitter for latest updates. I've also started posting more recent posts on my personal blog.

A while back I was working on a Django project and wanted to implement fast free text search. Instead of using a regular database for this search function — such as MySQL or PostgreSQL — I decided to use a NoSQL database. That is when I discovered ElasticSearch.

ElasticSearch indexes documents for your data instead of using data tables like a regular relational database does. This speeds up search, and offers a lot of other benefits that you don’t get with a regular database. I kept a regular relational database as well for storing user details, logins, and other data that ElasticSearch didn’t need to index.

After searching for a long time on how to properly implement ElasticSearch with Django, I didn’t really find any satisfying answers. Some guides or tutorials were convoluted and seemed to be taking unnecessary steps in order to index the data into ElasticSearch. There was quite a bit of information on how to perform searching, but not as much about how the indexing should be done. I felt like there must be a simpler solution out there, so I decided to give it a try myself.

I wanted to keep it as simple as possible, because simple solutions tend to be the best ones in my opinion. KISS (Keep It Simple Stupid), Less is More and all of that stuff is something that resonates with me a lot, especially when every other solution out there is complex. I decided to use Honza Král’s example in this video to have something to base my code on. I recommend watching it, although it is a bit outdated at this point.

Since I was using Django — which is written in Python — it was easy to interact with ElasticSearch. There are two client libraries to interact with ElasticSearch with Python. There’s elasticsearch-py, which is the official low-level client. And there’s elasticsearch-dsl, which is build upon the former but gives a higher-level abstraction with a bit less functionality.

We will get into some example soon, but first I need to clarify what we want to accomplish:

• Setting up ElasticSearch on our local machine and ensuring it works properly
• Setting up a new Django project
• Bulk indexing of data that is already in the database
• Indexing of every new instance that a user saves to the database
• A basic search example

All right, that seems simple enough. Lets get started by installing ElasticSearch on our machine. Also, all the code will be available on my GitHub so that you can easily follow the examples.

Installing ElasticSearch

Since ElasticSearch runs on Java you must ensure you have an updated JVM version. Check what version you have with java -version in the terminal. Then you run the following commands to create a new directory, download, extract and start ElasticSearch:

mkdir elasticsearch-example

wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.1.1.tar.gz

tar -xzf elasticsearch-5.1.1.tar.gz

./elasticsearch-5.1.1/bin/elasticsearch

When ElasticSearch starts up there should be a lot of output printed to the terminal window. To check that its up and running correctly open up a new terminal window and run this curl command:

curl -XGET http://localhost:9200

The response should be something like this:

{  "name" : "6xIrzqq",  "cluster_name" : "elasticsearch",  "cluster_uuid" : "eUH9REKyQOy4RKPzkuRI1g",  "version" : {    "number" : "5.1.1",    "build_hash" : "5395e21",    "build_date" : "2016-12-06T12:36:15.409Z",    "build_snapshot" : false,    "lucene_version" : "6.3.0"  },  "tagline" : "You Know, for Search"

Great, you now have ElasticSearch running on your local machine! It’s time to set up your Django project.

Setting up a Django project

First you create a virtual environment with virtualenv venv and enter it with source venv/bin/activate in order to keep everything contained. Then you install some packages:

pip install djangopip install elasticsearch-dsl

To start a new Django project you run:

django-admin startproject elasticsearchprojectcd elasticsearchprojectpython manage.py startapp elasticsearchapp

After you created your new Django projects you need to create a model that you will use. For this guide I chose to go with a good old fashioned blog post example. In models.py you place the following code:

from django.db import modelsfrom django.utils import timezonefrom django.contrib.auth.models import User# Create your models here.# Blogpost to be indexed into ElasticSearchclass BlogPost(models.Model):   author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='blogpost')   posted_date = models.DateField(default=timezone.now)   title = models.CharField(max_length=200)   text = models.TextField(max_length=1000)

Pretty straight forward, so far. Don’t forget to add elasticsearchapp to INSTALLED_APPS in settings.py and register your new BlogPost model in admin.py like this:

from django.contrib import adminfrom .models import BlogPost# Register your models here.# Need to register my BlogPost so it shows up in the adminadmin.site.register(BlogPost)

You must also python manage.py makemigrations, python manage.py migrate and python manage.py createsuperuser to create the database and an admin account. Now, python manage.py runserver, go to [http://localhost:8000/admin/](http://localhost:8000/admin/) and login. You should now be able to see your Blog posts model there. Go ahead and create your first blog post in the admin.

Congratulations, you now have a functioning Django project! It’s finally time to get into the fun stuff — connecting ElasticSearch.

Connecting ElasticSearch with Django

You begin by creating a new file called search.py in our elasticsearchapp directory. This is where the ElasticSearch code will live. The first thing you need to do here is to create a connection from your Django application to ElasticSearch. You do this in your search.py file:

from elasticsearch_dsl.connections import connectionsconnections.create_connection()

Now that you have a global connection to your ElasticSearch set-up you need to define what you want to index into it. Write this code:

from elasticsearch_dsl.connections import connectionsfrom elasticsearch_dsl import DocType, Text, Dateconnections.create_connection()class BlogPostIndex(DocType):    author = Text()    posted_date = Date()    title = Text()    text = Text()    class Meta:        index = 'blogpost-index'

It looks pretty similar to your model, right? The DocType works as a wrapper to enable you to write an index like a model, and the Text and Date are the fields so that they get the correct format when they get indexed.

Inside the Meta you tell ElasticSearch what you want the index to be named. This will be a point of reference for ElasticSearch so that it knows what index it’s dealing with when initializing it in the database and saving each new object instance created.

Now you need to actually create the mapping of your newly created BlogPostIndex in ElasticSearch. You can do this and also create a way to do the bulk indexing at the same time — how convenient right?

Bulk indexing of data

The bulk command is located in elasticsearch.helpers which is included when you installed elasticsearch_dsl since it is built on top of that library. Do the following in search.py:

...from elasticsearch.helpers import bulkfrom elasticsearch import Elasticsearchfrom . import models...

...def bulk_indexing():    BlogPostIndex.init()    es = Elasticsearch()    bulk(client=es, actions=(b.indexing() for b in models.BlogPost.objects.all().iterator()))

“What is going on here?” you might be thinking. It’s not that complicated, actually.

Since you only want to do bulk indexing whenever you change something in our model you init() the model which maps it into ElasticSearch. Then, you use the bulk and pass it an instance of Elasticsearch() which will create a connection to ElasticSearch. You then pass a generator to actions= and iterate over all the BlogPost objects you have in your regular database and call the .indexing() method on each object. Why a generator? Because if you had a lot of objects to iterate over a generator would not have to first load them into memory.

There is just one problem with the above code. You don’t have an .indexing() method on your model yet. Lets fix that:

...from .search import BlogPostIndex...

...# Add indexing method to BlogPostdef indexing(self):   obj = BlogPostIndex(      meta={'id': self.id},      author=self.author.username,      posted_date=self.posted_date,      title=self.title,      text=self.text   )   obj.save()   return obj.to_dict(include_meta=True)

You add the indexing method to the BlogPost model. It returns a BlogPostIndex and gets saved to ElasticSearch.

Lets try this out now and see if you can bulk index the blog post you previously created. By running python manage.py shell you go into the Django shell and import your search.py with from elasticsearchapp.search import * and then run bulk_indexing() to index all the blog posts in your database. To see if it worked you run the following curl command:

curl -XGET 'localhost:9200/blogpost-index/blog_post_index/1?pretty'

You should get back your first blog post in the terminal.

Indexing of newly saved instance

Next you need to add a signal that fires the .indexing() on each new instance that is saved every time a user saves a new blog post. In elasticsearchapp create a new file called signals.py and add this code:

from .models import BlogPostfrom django.db.models.signals import post_savefrom django.dispatch import receiver@receiver(post_save, sender=BlogPost)def index_post(sender, instance, **kwargs):    instance.indexing()

The post_save signal will ensure that the saved instance will get indexed with the .indexing() method after it is saved.

In order for this to work we also need to register Django that we’re using signals. We do this opening apps.py and adding the following code:

from django.apps import AppConfigclass ElasticsearchappConfig(AppConfig):    name = 'elasticsearchapp'    def ready(self):        import elasticsearchapp.signals

To to complete this we also need to tell Django that we’re using this new configuration. We do this inside the __init__.py inside our elasticsearchapp directory by adding:

default_app_config = 'elasticsearchapp.apps.ElasticsearchappConfig'

Now the post_save signal is registered with Django and is ready to listen for whenever a new blogpost is being saved.

Try it our by going into the Django admin again and saving a new blogpost. Then check with a curl command if it was successfully indexed into ElasticSearch.

Simple search

Now lets make a simple search function in search.py to find all posts filtered by author:

...from elasticsearch_dsl import DocType, Text, Date, Search...

...def search(author):    s = Search().filter('term', author=author)    response = s.execute()    return response

Lets try the search out. In the shell: from elasticsearchapp.search import * and run print(search(author="<author name&gt;")) :

>>> print(search(author="home"))<Response: [<Result(blogpost-index/blog_post_index/1): {'text': 'Hello world, this is my first blog post', 'title':...}>]>

There you have it! You have now successfully indexed all your instances into ElasticSearch, created a post_save signal that indexes each newly saved instance, and created a function to search our ElasticSearch database for your data.

Conclusion

This was a quite lengthy article but I hope it is written simple enough for even the beginner to be able to understand.

I explained how to connect a Django model to ElasticSearch for indexing and searching, but there is so much more that ElasticSearch can do. I recommend reading on their website and exploring what other possibilities exist, such as spatial operations and full text search with intelligent highlighting. Its a great tool and I will be sure to use it in future projects!

If you liked this article or have a comment or suggestion, please feel free to leave a message below. And stay tuned for more interesting stuff!