Elasticsearch API and Ingestion Pipeline

Elasticsearch API

Task: Create a new index containing three documents. After creation, delete one document using the DELETE method and another using a query-based deletion. Then, reindex the remaining documents and perform an index flush.

First, let's open Kibana Dev Tools and initiate the creation of a new index.

PUT /weinnovate 

Now, let's add three documents to this index:

POST /weinnovate/_doc/one
{
  "name": "Fares",
  "city": "Al Mahalla",
  "age": "23"
}

POST /weinnovate/_doc/two
{
  "name": "Omar",
  "city": "Cairo",
  "age": "23"
}

POST /weinnovate/_doc/three
{
  "name": "Ahmed",
  "city": "Mansoura",
  "age": "25"
}

Now, let's proceed with deleting the first document.

DELETE /weinnovate/_doc/one
GET /weinnovate/_search

Next, let's try delete one document by query.

POST /weinnovate/_delete_by_query
{
  "query":{
    "match":{
      "name": "Omar"
    }
  }
}

GET /weinnovate/_search

We now need to reindex the documents, which enables us to copy them from one index to another. Before reindexing, we need to create a new index

PUT weinnovate_new

Now, let's reindex the data from weinnovate to weinnovate_new.

POST _reindex
{
  "source": {
    "index": "weinnovate"
  },
  "dest": {
    "index": "weinnovate_new"
  }
}

GET /weinnovate_new/_search

Flushing an index in Elasticsearch ensures that all operations are written to disk. This can help free up memory and optimize performance.

POST weinnovate_new/_flush

Ingestion Pipline

Ingest pipelines let you perform common transformations on your data before indexing. For example, you can use pipelines to remove fields, extract values from text, and enrich your data.

Task: Create a pipeline with three distinct processors for a new index.

First, we will create a pipeline named weinnovate_pipeline, which will include three processors.

PUT _ingest/pipeline/weinnovate_pipline
{
  "description": "New Pipline for the weinnovate index",
  "processors": [
    {"append":{
      "field": "Status",
      "value": ["Active"]
      }
    }, 
    {"convert":{
      "field": "age",
      "type": "integer"
      }
    }, 
    {"uppercase":{
        "field": "name"
      }
    }
  ]
}

• The append processor adds a new value "Active" to the Status field.

• The convert processor converts the value of "age" to an integer.

• The uppercase processor converts the "name" field to uppercase.

We need to confirm that the pipeline was created successfully.

GET _ingest/pipeline/weinnovate_pipeline

Now, I want to apply this pipeline to the previously created "weinnovate" index. To do this, we first need to reindex the existing data while applying the pipeline during the reindexing process.

POST _reindex
{
  "source": {
    "index": "weinnovate"
  },
  "dest": {
    "index": "weinnovate_two",  
    "pipeline": "weinnovate_pipline"
  }
}

Let's verify this by retrieving the documents from the weinnovate_two index.

GET /weinnovate_two/_search

Alternatively, we can first create the pipeline and then apply it when creating a new index.