How to ETL with MongoDB & Postgres (Part 3)
Tags: programming, javascript, technology, softwaredevelopment
Published: 2019-02-03 22:23:41

---
title: How to ETL with MongoDB & Postgres (Part 3)
published: true
tags: programming,javascript,technology,software-development
canonical_url: https://medium.com/chingu/how-to-etl-with-mongodb-postgres-part-3-4aeb95c2cd3a
---

Part 3, Designing the Databases

![](https://cdn-images-1.medium.com/max/1024/1*vlsoaZWot_ig5SX2Px8Cvg.jpeg)<figcaption>Photo by <a href="https://unsplash.com/photos/T6fDN60bMWY?utm_source=unsplash&amp;utm_medium=referral&amp;utm_content=creditCopyText">Edho Pratama</a> on <a href="https://unsplash.com/search/photos/design?utm_source=unsplash&amp;utm_medium=referral&amp;utm_content=creditCopyText">Unsplash</a></figcaption>

The goal of an extract, transform, and load (ETL) process is to acquire source data, cleanse and organize it, and finally store it in a permanent location. Cleaning the data and organizing its structure is typically performed in a temporary staging area, while the permanent location is usually an operational database or data warehouse.

Two essential components of the ETL process are the path data takes through the app and the design of the databases used for the staging and operational databases. Factors influencing the design of these storage locations are how the information is to be accessed and manipulated, and the steps in the ETL process.

For example, the operational database must meet users’ performance and availability expectations. It is important that new data be quickly loaded, so there is no impact to user [service level agreements](https://en.wikipedia.org/wiki/Service-level_agreement) (SLA’s). If climate data is needed to support daily weather forecasts, it is imperative that data from the observation stations are available in time for the forecasts to be useful. After all, of what value is Tuesday’s forecast if it isn’t available until Wednesday?

In this article, we’ll define the requirements and constraints of the ETL procedure the Climate Explorer app depends on and more importantly, how it influences the design of both the staging and the operational databases.

Previous articles in this series proposed the Climate Explorer application as a means to explore an ETL process that uses MongoDB for the staging area and PostgreSQL for the operational database. They also described the design and implementation of Climate Explorer’s frontend and backend environments.

- [Part 1, Learning the Lay of the Land](https://dev.to/jdmedlock/how-to-etl-with-mongodb-and-postgres-part-1-5775))
- [Part 2, Design & Set Up the Environment](https://dev.to/jdmedlock/how-to-etl-with-mongodb--postgres-part-2-1422)

The [Climate Explorer](https://github.com/jdmedlock/climateexplorer/tree/feature/04-config-dbs) application for this article is located in the [feature/06-article-part3](https://github.com/jdmedlock/climateexplorer/tree/feature/06-article-part3) branch on GitHub.

### Designing the ETL Process

![](https://cdn-images-1.medium.com/max/1024/1*TRDOdBFGd73v2LKwJWePfQ.png)<figcaption>Figure 1 — Climate Explorer ETL Process</figcaption>

The overall goal of the ETL process is to get data from its source to an operational database. Each phase in the ETL process has a specific objective and set of constraints:

- _Extract_ — Obtain and stage daily climate observations from the NOAA site in a way that anticipates failure and supports automatic recovery.
- _Transform_ — Cleanse the data and organize it in a way that supports fast loading into the operational database.
- _Load_ — Populate the operational database from the staging area. Loading the operational database must be efficient, and it must support incremental loading since new weather observations arrive daily.

#### Extract Procedure

The extract procedure uses the [file transfer protocol](https://en.wikipedia.org/wiki/File_Transfer_Protocol) (FTP) to retrieve daily weather observation files from the NOAA site as follows:

![](https://cdn-images-1.medium.com/max/1024/1*kqkVaf-1xpPVAvKOclkRKQ.png)<figcaption>Figure 2 — ETL Extract Process</figcaption>

Extraction starts by retrieving the entry for the most recently completed file from the checkpoint. Checkpoint entries define the state of each file encountered during previous runs and are used to determine which file to start from in the current execution.

- file name
- modification date
- ETL state — extraction complete, transformation complete, and load complete

The starting point will be one of the following, in order of priority:

1. The most recent checkpoint entry that was started, but not completed. This is the case if all three ETL state flags are disabled.
2. A file name not previously encountered
3. A file whose modification date is more recent than its checkpoint entry.

For the initial load of the staging area, this process may use a configuration option to control the duration of the run based on either elapsed time or a file count. Once the initial data load has been completed this process may be scheduled to run as needed to add new or modified files created by observation stations.

The last step in the extract process is particularly important since it ensures that if the current run is interrupted the next run will repeat processing of the current file.

It is crucial to have a robust error handling strategy that considers edge cases. For example, failing to notify the DevOps team to take corrective action when errors are detected can result in undesirable side effects such as repeatedly processing the same file on each run.

#### Transformation Procedure

The transformation procedure is responsible for analyzing the raw data added to the staging area in the following steps:

![](https://cdn-images-1.medium.com/max/601/1*h4Sw8z0XRM-GRzWzQoTIbw.png)<figcaption>Figure 3 — ETL Transformation Process</figcaption>

The primary objective of the transformation procedure is not only to ensure that documents and fields are valid but to automatically correct errors whenever possible. This may include modifying source data, excluding documents when there is no corrective action, logging errors, and notifying the appropriate personnel in the event of a catastrophic failure.

This includes ensuring that relations between documents contain the correct shared field values on both sides of each relation in preparation for loading into the SQL operational database.

#### Load Procedure

The last step in the ETL process is to transfer the staged data to the operational database. It might sound like a simple task, but the operational database is a SQL DBMS, so data must be inserted in a manner that takes into account the relationships between tables.

Since SQL relationships are based on shared column values insertion of new data starts with parent table rows followed by the related child table rows. This can become difficult as the number and complexity of relationships between tables increases.

Consider the case shown in Figure 4. Since the _Region_ table is a parent of both the _Customer_ and _SalesRep_ tables, and since a one-to-many relationship exists between _Customer_ and _SalesRep_, related rows are added to the _Region_ table, then _SalesRep_, and finally _Customer_. Attempting to insert a _Customer_ row before its associated parent _Region_ or _SalesRep_ would result in the violation of a foreign key constraint.

![](https://cdn-images-1.medium.com/max/404/1*s1Atfp5D1SYZv5YIidaqiQ.png)<figcaption>Figure 4 — Example Application Database Design</figcaption>

Figure 5 presents a more complicated example where one of the tables, _Part_, participates in a nested relationship with itself. This kind of relationship is also known as a [bill of materials](https://en.wikipedia.org/wiki/Bill_of_materials) relationship. In other words, one where a single table contains multiple subcomponents made up of other subcomponents and parts.

![](https://cdn-images-1.medium.com/max/947/1*s9B8bkftryjz5CT1HpMI3w.png)<figcaption>Figure 5 — Example BOM Database Design</figcaption>

In this example, inserting rows into the _Part_ table may be performed in one of two ways.

The first is to group input records by product and component in a way that allows the relationship to be built from the “bottom up” to ensure that a child isn’t added before its parent. Parts with no subcomponents are loaded first, then subcomponents that are parents of parts, and so on up the relationship chain.

![](https://cdn-images-1.medium.com/max/447/1*TM5nH5nl4wpQIV3AzFsr9g.png)<figcaption>Figure 6 — Parts table insertion order for BOM</figcaption>

The second option is to remove the relationship on the _Parts_ table, insert new rows, and then re-establish the relationship. Although viable for an initial data load the fact that downtime is required and thus impacts the user makes it an option of last resort.

Luckily, the entity-relationships in Climate Explorer are relatively straightforward, so this situation isn’t applicable.

### Designing the Staging Database

![](https://cdn-images-1.medium.com/max/922/1*dpx6cHgPdKwF_LFYqKJoHA.png)<figcaption>Figure 7 — Staging Database ER Diagram</figcaption>

Climate Explorer’s [staging database](https://en.wikipedia.org/wiki/Staging_(data)) uses a set of MongoDB collections with imported file types mapped to unique collections. For example, ghcnd-stations.txt is loaded into the Station collection. Figure 7 shows the staging database collections and their logical relationships.

The MongoDB archive containing the staging database definitions is located [here](https://github.com/jdmedlock/climateexplorer/blob/feature/06-article-part3/server/src/artifacts/mongodb/climateexplorer_mongodump_archive_20190201_085700) in the Climate Explorer [repo](https://github.com/jdmedlock/climateexplorer/tree/feature/06-article-part3) on Github.

#### Database Normalization

The daily files containing weather data from stations includes an array of up to 31 observations, one for each day of the month. Data added into the staging database observations is split across two collections — the parent identifying the observations station, year, and month and its children for each daily observation.

Collections are also defined for the codes used in the _Station_, _Observation_, and _DailyWeather_ documents. Rather than relying on documentation in `readme.txt`, as is the case in the NOAA files, lookup tables provide authoritative definitions for encoded fields. For example, the _Network_ table defines the network codes used in the _Station_ table.

Nothing prohibits importing information into the staging area without modification. Instead, we’ve opted to normalize the staged data to simplify the process of loading data into the operational database in the ETL load phase.

See the section “Designing the Operational Database” below for more information about normalization.

#### Schema Validation

![](https://cdn-images-1.medium.com/max/821/1*tIchb_ObD_kB6Ge1npa2Ug.png)<figcaption>Figure 8 — Schema Validator for Country Collection</figcaption>

Document updates and insertions are validated using MongoDB [schema validators](https://docs.mongodb.com/manual/core/schema-validation/index.html) defined on each collection. Validators in Climate Explorer collections only perform field type checking. Analysis of field content is left to the transformation logic since the goal is to limit the number of rejected records by automatically correcting errors.

#### Relations, Embedded Documents, & Indexes

Since the state of the NOAA data is unknown at this time, we’ve chosen not to use relations or embedded documents in the staging database. The data must first be analyzed and cleansed to ensure the validity of the logical relationships.

Indexes are defined for all document fields on the “many” side of a relation, and on the “one” side based on document volume as is the case for the _Station_ and _Observation_ documents.

#### Extra Document Fields

The extraction process relies on MongoDB transactions to commit or rollback a file in its entirety based on whether or not an error occurred. Since there is no use case for record level recovery/restart document fields defining where data is imported from or when the import occurred are needed.

### Designing the Operational Database

![](https://cdn-images-1.medium.com/max/957/1*jdzzKE_lNeofhApBN2oW0Q.png)<figcaption>Figure 9 — Operational Database DB Diagram</figcaption>

The responsibility of the [operational database](https://en.wikipedia.org/wiki/Operational_database) is to support Climate Explorer’s ability to satisfy user queries. This database is the permanent home for climate data and is implemented as a set of PostgreSQL tables and relationships.

The PostgreSQL script that defines the operational database is located [here](https://github.com/jdmedlock/climateexplorer/tree/feature/06-article-part3/server/src/artifacts/postgresql) in the Climate Explorer [repo](https://github.com/jdmedlock/climateexplorer/tree/feature/06-article-part3) on Github.

#### Updates & Indexes

Updates to climate data are prohibited since the [system of record](https://en.wikipedia.org/wiki/System_of_record) is the NOAA Global Historical Climatology Network. Due to this the operational database is designed for fast queries rather than updates.

Indexes are defined on both parent and child columns in a relation. No additional indexes have been created, but new ones will be added later as dictated by performance requirements.

#### Database Normalization

Much of the [database normalization](https://en.wikipedia.org/wiki/Database_normalization) process was performed during the design of the staging database. This is driven by the assumption that minimizing the load time of the operational database is a higher priority than reducing the load time of the staging database.

### Retrospective — Failing Forward

#### Retrieving Files Using FTP

As previously mentioned the extract process uses the file transfer protocol (FTP) to retrieve files from the NOAA site. The NPM package [promise-ftp](https://www.npmjs.com/package/promise-ftp) provides an interface to the FTP protocol, and the middleware module FTPAPI.js provides an interface between it and the app.

Establishing a connection to the source requires that an FTP session is created from dataSources in the Apollo context object as shown below.

```
// Data sources required by the resolvers. These are available to 
// subclasses of DataSource via config.context.
const ftpSession = new FTPAPI( {
  host\_url: process.env.NOAA\_FTP\_URL,
  host\_port: process.env.NOAA\_FTP\_PORT,
  user: process.env.NOAA\_FTP\_USER,
  password: process.env.NOAA\_FTP\_PASSWORD
} );

const mongo = new MongoAPI();
const postgres = new PostgresAPI();
const country = new Country(mongo);
const location = new Location(postgres);
const user = new User(mongo);

const dataSources = () => ({
  ftpSession: ftpSession,
  country : country,
  location: location,
  user: user,
});

.
    .
    .

// Create and start the Apollo Server
const server = new ApolloServer({
  typeDefs,
  dataSources,
  context,
  resolvers,
});
```

The following example from Countries.js demonstrates the retrieval of files from the source.

```
async extractCountriesFromGhcnd() {
  const ftpSession = this.context.dataSources.ftpSession;
  return new Promise(function (resolve, reject) {
    ftpSession.connect()
    .then(() => {
      return ftpSession.getFile(`${process.env.NOAA_FTP_GHCN_DIRECTORY}/${process.env.NOAA_FTP_COUNTIES_FILE}`
      );
    })
    .then(stream => {
      let countries = '';
      stream.on('readable', function() {
        let data;
        while (data = this.read()) {
          countries += data.toString();
        }
        resolve(countries);
        ftpSession.disconnect();
      });
    })
    .catch(err => {
      throw new Error('FTP from NOAA failed.')
    });
  })
  .then(countries => {
    const countriesObject = this.convertToObject(countries);
    return countriesObject;
  })
  .catch(err => {
    throw new Error('FTP from NOAA failed.');
  });
}
```

#### Cleaning Up the Staging Database

When should data be removed from the staging area? There is no reason to retain data in the staging database once it is successfully added to the operational database. Two situations influence this:

- _Load failure_. Wrapping a load of each file in a SQL transaction forces all inserts to be rolled back if the load fails, thus enforcing the requirement that all insertions must be successful before any can be committed.
- _Bad Data_. Even though loading a file may be successful it’s possible that an integrity issue may be discovered later. It’s safe to assume that the probability of this occurring will diminish over time.

Based on this staging data will be retained until the start of the next load run.

### Wrapping It Up

![](https://cdn-images-1.medium.com/max/1024/1*143qCDpww31xSFrimYJlqg.jpeg)<figcaption>Photo by <a href="https://unsplash.com/photos/Wpnoqo2plFA?utm_source=unsplash&amp;utm_medium=referral&amp;utm_content=creditCopyText">Mika Baumeister</a> on <a href="https://unsplash.com/search/photos/data-diagram?utm_source=unsplash&amp;utm_medium=referral&amp;utm_content=creditCopyText">Unsplash</a></figcaption>

The purpose of database design is to organize data in a way that makes it easy to access and manipulate. Developing a solid database design requires matching application requirements and data flow with the capabilities provided by the DBMS. This approach has been followed in this article to complete the design of the staging and operational databases.

As dictated by an Agile approach, known requirements have been taken into consideration, and assumptions and “blue sky” ideas that have yet to materialize as requirements have been documented, but not acted on. This means revisions will become necessary if changes arise to existing requirements or new ones surface.

We’ll undoubtedly encounter this in the next article where we’ll implement the ETL process. Get ready for some code!

**_Up Next — Part 4, Extracting, Transforming, & Loading Data_**

* * *