Part 6: Set up Elasticsearch for data transformation and data ingestion
Tags: elasticsearch, data, beginners, webdev
Published: 2022-09-07 15:14:45

[Table of Content](https://dev.to/lisahjung/beginners-guide-to-building-a-full-stack-app-nodejs-react-with-elasticsearch-5347) | [Read Next: Part 7: Set up the Node.js server to retrieve API data and send the data to Elasticsearch](https://dev.to/lisahjung/part-7-set-up-the-nodejs-server-to-retrieve-api-data-and-send-the-data-to-elasticsearch-50fm)

###Resources
*Would you rather watch a video to learn this content? Click on the link below!*

- [Episode 6: Set up Elasticsearch for data transformation and data ingestion](https://www.youtube.com/watch?v=pM0ZN989E4s)

### Set up Elasticsearch for data transformation and data ingestion
In [part 5](https://dev.to/lisahjung/part-5-plan-for-efficient-data-storage-and-search-performance-in-elasticsearch-22mo), we figured out:

1. a plan for transforming data before ingesting data into Elasticsearch
2. a desired mapping for our data

In this blog, we will accomplish two tasks:

1. create an `ingest pipeline` to transform the retrieved data
2. create an index called `earthquakes` with the desired mapping

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ql0k9qnkcy1ppolesqcb.png)
   
Before we get started, let’s talk about the data journey for our app.

We are building an app where users can search for earthquake data stored in Elasticsearch.

In [part 7](https://dev.to/lisahjung/part-7-set-up-the-nodejs-server-to-retrieve-api-data-and-send-the-data-to-elasticsearch-50fm), we will set up our server to retrieve the data from the USGS API and send the data to Elasticsearch `ingest pipeline`. 

![ezgif com-gif-maker (10)](https://user-images.githubusercontent.com/60980933/186692850-fdf7c617-2038-4aa5-88b7-6c0f40f1412d.gif)

`Ingest pipeline` is used for data transformation.

It consists of a series of configurable tasks called processors. Each processor performs a specialized task. For example, it can remove fields, extract values from text, enrich your data and etc. 

![ezgif com-gif-maker (11)](https://user-images.githubusercontent.com/60980933/186693245-0a258378-c424-4b57-be82-7071bf23c7ce.gif)

Each processor runs in the order you set them up and they make specific changes to the incoming documents.

After the processors have run, Elasticsearch will add the transformed documents to the earthquake index we will create. 

![ezgif com-gif-maker (12)](https://user-images.githubusercontent.com/60980933/186693539-369b9ec0-7161-46da-bff2-9128c8375da4.gif)

If you want to delve deeper into `ingest pipelines`, check out this [documentation](https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html)!

###Set up Elasticsearch for data transformation and data ingestion

####Step 1: Review the data transformation requirements

Our `ingest pipeline` will be used to transform the data retrieved from the USGS API.

Before we create an `ingest pipeline`, let's review what changes we want to make to the data.

1. remove the unnecessary info from the retrieved data 
2. change the Unix epoch time in the field `time` to human readable timestamp
3. create fields `coordinates.lat` and `coordinates.lon` as shown below
![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/92y141tk24ny2dcn3a86.png) 

####Step 2: Create an `ingest pipeline`

Ingest pipelines can be created and managed via Kibana's `Ingest Pipelines feature` or the `ingest APIs`.

We will be using Kibana to create this pipeline. 

*From the Kibana home page, click on the `Stack Management` option(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/366x83egdc1aqh53r6n7.png)
  
*From the `Stack Management page`, click on the `Ingest Pipelines` option(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/pge5gykzb1u8n2ubp23i.png)
  
*Click on the `Create pipeline` option(red box) and select the `New pipeline` option from the drop down menu(blue box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jw3ph9s6xlpo8oinj240.png)
  
*Name your pipeline to whatever it makes sense to you.* 

For this project, I named mine `earthquake_data_pipeline`(red box).

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/p5xs3l7jqw5wrd5nemj8.png)
  
####Step 3: Add the desired processors to the pipeline

*Click on the `Add a processor` option(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/0axc99w0l3zggfb2iozd.png)
   
You should see the following pop up menu. 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jsp65etfb3l9ydy1kzcv.png)
 
*Task 1: Remove the fields that we do not need from the retrieved data*

Here is an example of an earthquake object from the USGS earthquake API:
```javascript
   {
     "type":"Feature",
     "properties":{
        "mag":1.13,
        "place":"11km ENE of Coachella, CA",
        "time":1650316843970,
        "updated":1650317059011,
        "tz":null, 
        "url":"https://earthquake.usgs.gov/earthquakes/eventpage/ci40240408",
        "detail":"https://earthquake.usgs.gov/earthquakes/feed/v1.0/detail/ci40240408.geojson",                
        "felt":null,
        "cdi":null,
        "mmi":null,
        "alert":null,
        "status":"automatic",
        "tsunami":0,
        "sig":20,
        "net":"ci",
        "code":"40240408",
        "ids":",ci40240408,",
        "sources":",ci,",
        "types":",nearby-cities,origin,phase-data,scitech-link,",
        "nst":37,
        "dmin":0.07687,
        "rms":0.26,
        "gap":48,
        "magType":"ml",
        "type":"earthquake",
        "title":"M 1.1 - 11km ENE of Coachella, CA"
     },
     "geometry":{
        "type":"Point",
        "coordinates":[
          -116.0736667,
          33.7276667,
          2.09
        ]
      
    },
     "id":"ci40240408"
   }
```
The following is a sample document with the desired fields we want to store in Elasticsearch.

```javascript
{
  "mag": 1.13,
  "place": "11km ENE of Coachella, CA",
  "time": 2022-05-02T20:07:53.266Z,
  "url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci40240408",
  "sig": 20,
  "type": "earthquake",
  "depth": 2.09,
  "coordinates": {
    "lat": 33.7276667,
    "lon": -116.0736667
  }
}
```
As you can see, API earthquake object has additional info we do not need.

By comparing the two, we can identify the fields that we want to remove from the retrieved data. 

Here is the list of fields that we do not need. 

1. `updated`
2. `tz`
3. `detail`
4. `felt`
5. `cdi`
6. `mmi`
7. `alert`
8. `status`
9. `tsunami`
10. `net`
11. `code`
12. `ids`
13. `sources`
14. `types`
15. `nst`
16. `dmin`
17. `rms`
18. `gap`
19. `magType`
20. `title`

To remove these fields, we can use the `Remove` processor. 

*Under the processor section, type `Remove` in the search bar. Click on the `Remove` processor(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/sdt7vielavyziwq4vbn7.png)
  
Elasticsearch will display the following menu. In the `Fields` section(red box), we will add the names of the fields that we wish to remove. 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/x7m88jq0wa7y8m9aabvs.png)
 
*Type in the name of the field you wish to remove then press enter. You will see that the field name has been added.* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/dexi9i91r1s5p63m86ba.png)
  
*In the same box, repeat the same process to specify the names of all the fields we wish to remove.* 

Your screen should look like the image below(red box). 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/w53j9bmurw6gp2ctd21e.png)

*Activate the `Ignore missing` option(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/u34q5hsggzq7ao06r8r0.png)

*Click on the `Add` button(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/h2xvhagsqxn2e24l8jgj.png)
 
You will see that the `Remove` processor has been added to the `earthquake_data_pipeline`(red box). 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ujcqmtv878ppjnvgoxw4.png)
         
*Task 2: Change the Unix epoch time in the field `time` to human readable timestamp*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hbcxim21oiw4lsf8ge41.png)
 
In order to make this change, you should use the `date` processor. 

The `date` processor converts time from one format to another. 

*Click on the `Add a processor` option(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/l3vn77tasc9s04zf4pmz.png)
  
*Under the `Processor` section, type in `Date` and click on the `Date` option from the drop down menu(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/e9t3jbgibogon9j3zoax.png)
  
You should see the following drop down menu. 

*In the `Field` section(red box), type the name of the field we wish to convert (`time`).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hgxvvcvh8luutvz65d6b.png)
  
In the `Formats` section(red box), we will specify the desired date formats. The format shown in the results card is called `UNIX_MS`. 

*Type it into this section and hit enter. Then, click on the `Add` button(green box) to add the `date` processor to the `ingest pipeline`.*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/8dfmvtqisrbuhbqzdypx.png)
  
You will see that the `Date` processor has been added to the `earthquake_data_pipeline`(red box).

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/8rqpjqu673ap4twd94m8.png)
   
When the data goes through the `date` processor, the content of the field `time` will be converted to the `UNIX_MS` format then stored in a *new field* called `@timestamp`. 

After this process is finished, we do not need the original field `time`. Therefore, we will remove the field `time` after the data goes through the `date` processor.

*From the `Create pipeline` page, click on the `Add a processor` option(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/klqf0bgxxm5eajde8i8l.png)
 
*Under the `Processor` section, type in `Remove` and hit enter(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wmjg3aetsj8385cjla8r.png)
 
*Under the `Fields` section, type in `time` and hit enter(red box). Activate the `ignore missing` option(blue box). Then, click on the `Add` button(green box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ftw194tbpyed1jpvju74.png)
  
You will see that the `Remove` processor for the field `time` has been added(red box). 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/cr73j2gotzbun9rmi62d.png)
  
*Task 3: Create fields called `coordinates.lat` and `coordinates.lon`*

*From the same page, click on the `Add a processor` option(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/nrqwdwwkizvazzgc7qaq.png) 

*Under the `processor` section, type in `Rename`  and hit enter(red box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/nao1uvhqxve1lcm81tvd.png) 

*Under the `Field` section, type in `latitude`(red box). Under the `Target field` section, type in `coordinates.lat`(blue box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hhyivebuhkwmnj5b8lcs.png)

This step will rename the field `latitude` in the incoming data to `coordinates.lat`. 

*Activate the `Ignore missing` option(yellow box) then click on the `Add` button(green box) to add this processor to the `earthquake_data_pipeline`.* 
   
You will see that the `Rename` processor for the field `latitude` has been added(red box).

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/lgt50a3k604w7rl9prbh.png)
  
Next, we will repeat the same process to add a `Rename` processor to rename the field `longitude` from the incoming data to `coordinates.lon`. 

This process is identical to the steps you have performed for the field `latitude`. 
 
Once this process is done, you will see that a `Rename` processor for the field `longitude` has been added to the `earthquake_data_pipeline`(red box). 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/810qpxlma5t1mw8t35td.png)
    
We have added all the necessary processors to transform our data. 

Before creating the `earthquake_data_pipeline`, make sure the order of the processors are listed in the order you want them to run.

These processors run sequentially! 

*Why does the order matter?*

Let's say you accidentally reversed the order of `date` processor for the field `time` with the `Remove` processor for the field `time`.

When the data goes through the `ingest pipeline`, the data will go through the `remove` processor before the `date` processor. As a result, the field `time` will be removed before it gets to the `date` processor. 

The `date` processor will have nothing to work with! 

This is why we double check the order of the processors before creating the `ingest_pipeline`!

*Next, we will create the `earthquake_data_pipeline` by clicking on the `Create pipeline` button(blue box).* 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/gnlqxy79jgwx30tszgsa.png)
  
You will see that the `earthquake_data_pipeline`(red box) has been created. If you scroll down on the `Processors` section(blue box), you will see the list of processors we have added to this pipeline. 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/a116m3mkpv38aapn6i6f.png)
    
####Step 4: Create an index called `earthquakes` with the desired mapping

We will accomplish this step using `Kibana Dev Tools`. 
 
*Click on the `menu` icon(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/fwkjv3w85pxatkipnqv4.png)
  
*Scroll down on the drop down menu and click on `Dev Tools`(red box).*

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wpe4zhhcn6plxsuts94k.png)
  
You should see `Dev Tools`(also known as Kibana console) on the screen. 

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5xq6hqvndv9sarubeyqw.png)

*In the left panel of the Kibana console, copy and paste the following.*
```javascript
PUT earthquakes
{
  "mappings": {
    "properties": {
      "@timestamp": {
        "type": "date"
      },
      "coordinates": {
        "type": "geo_point"
      },
      "depth": {
        "type": "float"
      },
      "mag": {
        "type": "float"
      },
      "place": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword"
          }
        }
      },
      "sig": {
        "type": "short"
      },
      "type": {
        "type": "keyword"
      },
      "url": {
        "enabled": false
      }
    }
  }
}
```
Your Kibana console should look the following:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/lr8j11miivguhh13ssyq.png)
   
*Click on the green arrow to send the request.* 

Expected output from Elasticsearch:
Elasticsearch will create an index called `earthquakes` with the desired mapping we defined above!

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/kdj6lp3xtbw902mxd29g.png)
 
**Summary**

In this blog, we have created:
1. an `ingest pipeline`(earthquake_data_pipeline) to transform the retrieved data from the USGS API
2. an index called `earthquakes` with the desired mapping

In the next blog, we will set up the server to retrieve earthquake data from the USGS API and send the data to the `earthquake_data_pipeline` in Elasticsearch.

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/n6yys382n2bh3su316tw.png)
 
Once the data transformation is complete, the transformed data will be ingested into the `earthquakes` index.

Move on to [Part 7](https://dev.to/lisahjung/part-7-set-up-the-nodejs-server-to-retrieve-api-data-and-send-the-data-to-elasticsearch-50fm) to set up the Node.js server to retrieve API data and send the data to Elasticsearch for data transformation and data ingestion!