How to use streams to ETL data?

Streams are a built-in feature in Node.js and represent asynchronous flow of data. Streams are also a way to handle reading and/or writing files. A Node.js stream can help process large files, larger than the free memory of your computer, since it processes the data in small chunks.

Streams in Node.js

This is the fifth article of a series about streams in Node.js. This article is about how to perform ETL operations (Extract, Transform, Load) on CSV data using streams.

Streams in Node.js

• What is a Stream in Node.js?
• Connect streams with the pipe method
• Handle stream errors
• Connect streams with the pipeline method
• Use streams to extract, transform and load data (this article)

Overview

When working with a flat data, we can just use the fs module and streams to process the data (memory-efficient). Instead of reading all the data into memory, we can read it in small chunks with the help of streams to avoid overconsumption of the memory.

In this article we are going to create sample data in a CSV file, extract this data, transform it and load the data.

A C omma- S eparated V alues file is a delimited text file that uses a comma to separate values. Read more here. We will transform the CSV data to JSON or better to ndjson, which is basically a file of JSON records separated by newlines and with the file extension .ndjson. For sure, you are asking yourself - why are we not just using JSON? The main reason is fault tolerance. If only one single invalid record is written to JSON, the entire JSON file will be corrupted. The main difference between JSON and ndjson is that in ndjson files each line of a file must contain a single JSON record. Hence, a ndjson file contains valid JSON, but ndjson is not a valid JSON document. The ndjson format works well with streaming data and large sets of data, where each record is processed individually.

We are going to:

• 1. Create CSV sample data
• 1. Initialize project for NPM
• 1. Create a CSV parser
• 1. Add Transform stream
• 1. Run & Done

1. Create CSV data

Let's create some sample CSV data, you can use the sample data below, or create your own data with FakerJS and convert it to CSV.

id,firstName,lastName,email,email2,randomized
100,Jobi,Taam,[e-mail],[e-mail],Z lsmDLjL
101,Dacia,Elephus,[e-mail],[e-mail],Za jfPaJof
102,Arlina,Bibi,[e-mail],[e-mail],zmzlfER
103,Lindie,Torray,[e-mail],[e-mail],ibVggFEh
104,Modestia,Leonard,[e-mail],[e-mail]," Tit KCrdh"
105,Karlee,Cornelia,[e-mail],[e-mail],PkQCUXzq
106,Netty,Travax,[e-mail],[e-mail],psJKWDBrXm
107,Dede,Romelda,[e-mail],[e-mail],heUrfT
108,Sissy,Crudden,[e-mail],[e-mail],cDJxC
109,Sherrie,Sekofski,[e-mail],[e-mail],dvYHUJ
110,Sarette,Maryanne,[e-mail],[e-mail],rskGIJNF
111,Selia,Waite,[e-mail],[e-mail],DOPBe
112,Karly,Tjon,[e-mail],[e-mail],zzef nCMVL
113,Sherrie,Berriman,[e-mail],[e-mail],rQqmjw
114,Nadine,Greenwald,[e-mail],[e-mail],JZsmKafeIf
115,Antonietta,Gino,[e-mail],[e-mail],IyuCBqwlj
116,June,Dorothy,[e-mail],[e-mail],vyCTyOjt
117,Belva,Merriott,[e-mail],[e-mail],MwwiGEjDfR
118,Robinia,Hollingsworth,[e-mail],[e-mail],wCaIu
119,Dorthy,Pozzy,[e-mail],[e-mail],fmWOUCIM
120,Barbi,Buffum,[e-mail],[e-mail],VOZEKSqrZa
121,Priscilla,Hourigan,[e-mail],[e-mail],XouVGeWwJ
122,Tarra,Hunfredo,[e-mail],[e-mail],NVzIduxd
123,Madalyn,Westphal,[e-mail],[e-mail],XIDAOx
124,Ruthe,McAdams,[e-mail],[e-mail],iwVelLKZH
125,Maryellen,Brotherson,[e-mail],[e-mail],nfoiVBjjqw
126,Shirlee,Mike,[e-mail],[e-mail],MnTkBSFDfo
127,Orsola,Giule,[e-mail],[e-mail],VPrfEYJi
128,Linzy,Bennie,[e-mail],[e-mail],ZHctp
129,Vanessa,Cohdwell,[e-mail],[e-mail],RvUcbJihHf
130,Jaclyn,Salvidor,[e-mail],[e-mail],gbbIxz
131,Mildrid,Pettiford,[e-mail],[e-mail],snyeV
132,Carol-Jean,Eliathas,[e-mail],[e-mail],EAAjYHiij
133,Susette,Ogren,[e-mail],[e-mail]," BhYgr"
134,Farrah,Suanne,[e-mail],[e-mail],hYZbZIc
135,Cissiee,Idelia,[e-mail],[e-mail],PNuxbvjx
136,Alleen,Clara,[e-mail],[e-mail],YkonJWtV
137,Merry,Letsou,[e-mail],[e-mail],sLfCumcwco
138,Fanny,Clywd,[e-mail],[e-mail],Go kx
139,Trixi,Pascia,[e-mail],[e-mail],lipLcqRAHr
140,Sandie,Quinn,[e-mail],[e-mail],KrGazhI
141,Dania,Wenda,[e-mail],[e-mail],CXzs kDv
142,Kellen,Vivle,[e-mail],[e-mail],RrKPYqq
143,Jany,Whittaker,[e-mail],[e-mail],XAIufn
144,Lusa,Fillbert,[e-mail],[e-mail],FBFQnPm
145,Farrah,Edee,[e-mail],[e-mail],TrCwKb
146,Felice,Peonir,[e-mail],[e-mail],YtVZywf
147,Starla,Juan,[e-mail],[e-mail],aUTvjVNyw
148,Briney,Elvyn,[e-mail],[e-mail],tCEvgeUbwF
149,Marcelline,Ricarda,[e-mail],[e-mail],sDwIlLckbd
150,Mureil,Rubie,[e-mail],[e-mail],HbcfbKd
151,Nollie,Dudley,[e-mail],[e-mail],EzjjrNwVUm
152,Yolane,Melony,[e-mail],[e-mail],wfqSgpgL
153,Brena,Reidar,[e-mail],[e-mail],iTlvaS
154,Glenda,Sabella,[e-mail],[e-mail],zzaWxeI
155,Paola,Virgin,[e-mail],[e-mail],gJO hXTWZl
156,Aryn,Erich,[e-mail],[e-mail],qUoLwH
157,Tiffie,Borrell,[e-mail],[e-mail],cIYuVMHwF
158,Anestassia,Daniele,[e-mail],[e-mail],JsDbQbc
159,Ira,Glovsky,[e-mail],[e-mail],zKITnYXyhC
160,Sara-Ann,Dannye,[e-mail],[e-mail],wPClmU
161,Modestia,Zina,[e-mail],[e-mail],YRwcMqPK
162,Kelly,Poll,[e-mail],[e-mail],zgklmO
163,Ernesta,Swanhildas,[e-mail],[e-mail],tWafP
164,Giustina,Erminia,[e-mail],[e-mail],XgOKKAps
165,Jerry,Kravits,[e-mail],[e-mail],olzBzS
166,Magdalena,Khorma,[e-mail],[e-mail],BBKPB
167,Lory,Pacorro,[e-mail],[e-mail],YmWQB
168,Carilyn,Ethban,[e-mail],[e-mail],KUXenrJh
169,Tierney,Swigart,[e-mail],[e-mail],iQCQJ
170,Beverley,Stacy,[e-mail],[e-mail],NMrS Zpa f
171,Ida,Dex,[e-mail],[e-mail],hiIgOCxNg
172,Sam,Hieronymus,[e-mail],[e-mail],dLSkVe
173,Lonnie,Colyer,[e-mail],[e-mail],ZeDosRy
174,Rori,Ethban,[e-mail],[e-mail],SXFZQmX
175,Lelah,Niles,[e-mail],[e-mail],NwxvCXeszl
176,Kathi,Hepsibah,[e-mail],[e-mail],SOcAOSn
177,Dominga,Cyrie,[e-mail],[e-mail],IkjDyuqK
178,Pearline,Bakerman,[e-mail],[e-mail],vHVCkQ
179,Selma,Gillan,[e-mail],[e-mail],hSZgpBNsw
180,Bernardine,Muriel,[e-mail],[e-mail],AnSDTDa U
181,Ermengarde,Hollingsworth,[e-mail],[e-mail],IYQZ Nmv
182,Marguerite,Newell,[e-mail],[e-mail],kSaD uaHH
183,Albertina,Nisbet,[e-mail],[e-mail],Y jHyluB
184,Chere,Torray,[e-mail],[e-mail],loElYdo
185,Vevay,O'Neill,Vevay.O'[e-mail],Vevay.O'[e-mail],uLZSdatVn
186,Ann-Marie,Gladstone,[e-mail],[e-mail],fwKlEksI
187,Donnie,Lymann,[e-mail],[e-mail],deBrqXyyjf
188,Myriam,Posner,[e-mail],[e-mail],gEMZo
189,Dale,Pitt,[e-mail],[e-mail],OeMdG
190,Cindelyn,Thornburg,[e-mail],[e-mail],kvhFmKGoMZ
191,Maisey,Hertzfeld,[e-mail],[e-mail],OajjJ
192,Corina,Heisel,[e-mail],[e-mail],luoDJeHo
193,Susette,Marcellus,[e-mail],[e-mail],AXHtR AyV
194,Lanae,Sekofski,[e-mail],[e-mail],FgToedU
195,Linet,Beebe,[e-mail],[e-mail],DYGfRP
196,Emilia,Screens,[e-mail],[e-mail],LXUcleSs
197,Tierney,Avi,[e-mail],[e-mail],VegzbHH
198,Pollyanna,Thar,[e-mail],[e-mail],GjYeEGK
199,Darci,Elephus,[e-mail],[e-mail],DaQNdN

Create a project folder:

mkdir node-streams-etl

Create a csv file in the folder:

cd node-streams-etl
touch sample-data.csv

Copy all sample data into the csv file and save it. Use copy+paste or fs.writeFile in the REPL or with the -p flag in the terminal.

2. Initialize project for NPM

We are going to use npm packages, hence, we have to initialize the project to get a package.json

npm init -y

Let's add a main file for the code.

touch index.js

First, we are going to create a readable stream to read the CSV data from sample-date.csv, and a writable stream, which will be the destination. For now, we just copy the sample data. To connect readStream and writeStream we are going to use the pipeline method. Error handling is much easier than with the pipe method. Check out the article How to Connect streams with the pipeline method.

const fs = require('fs');
const { pipeline } = require('stream');

const inputStream = fs.createReadStream('data/sample-data.csv');
const outputStream = fs.createWriteStream('data/sample-data.ndjson');

pipeline(inputStream, outputStream, err => {
  if (err) {
    console.log('Pipeline encountered an error.', err);
  } else {
    console.log('Pipeline completed successfully.');
  }
});

3. Create a CSV parser

We have to convert the CSV file to JSON, as so often, for every problem, there is a package. In that use-case, there is csvtojson. This module will parse the header row to get key and then parse each row to create a JSON object.

Let's install it.

npm install csvtojson

After the successful installation we can require the module and add it to the pipeline after the inputStream. The data will flow from CSV file to CSV Parser then into Output file.

We are going to use the pipeline method, since it's the preferred way since Node.js v.10 to connect streams and pipe data between them. It also helps to clean up streams on completion or failure, because when an error occurs the streams involved will be destroyed to avoid memory leaks.

const fs = require('fs');
const { pipeline } = require('stream');
const csv = require('csvtojson');

const inputStream = fs.createReadStream('data/sample-data.csv');
const outputStream = fs.createWriteStream('data/sample-data.ndjson');

const csvParser = csv();

pipeline(inputStream, csvParser, outputStream, err => {
  if (err) {
    console.log('Pipeline encountered an error.', err);
  } else {
    console.log('Pipeline completed successfully.');
  }
});

4. Add Transform stream

The data is now emitted to the outputStream as ndjson with each data row a valid JSON. Now, we want to transform the data. Since we are using csvtojson, we could utilize the built-in subscribe method, which could be used to handle each record after it has been parsed. Though, we want to create a transform stream. Our sample data has the keys id, firstName, lastName, email, email2, randomized. We want to get rid of the randomized property in each entry and rename email2 to emailBusiness.

Transform streams must implement a transform method that receives chunk of data as the first argument. It will also receive the encoding type of the data chunk, and a callback function.

const transformStream = new Transform({
  transform(chunk, encoding, cb) {
    try {
      // clone person object
      let person = Object.assign({}, JSON.parse(chunk));
      // remove randomized property and rename email2 to emailBusiness
      person = {
        id: person.id,
        firstName: person.firstName,
        lastName: person.lastName,
        emailBusiness: person.email2,
      };
      cb(null, JSON.stringify(person) + `\n`);
    } catch (err) {
      cb(err);
    }
  },
});

Now let's add the transformStream to the pipeline.

pipeline(
  inputStream,
  csvParser,
  transformStream,
  outputStream,
  err => {
    if (err) {
      console.log('Pipeline encountered an error.', err);
    } else {
      console.log('Pipeline completed successfully.');
    }
  },
);

5. Run & Done

Run the application with node index.js and the data in the ndjson file should look like this.

{"id":"100","firstName":"Jobi","lastName":"Taam","emailBusiness":"[e-mail]"}
{"id":"101","firstName":"Dacia","lastName":"Elephus","emailBusiness":"[e-mail]"}
{"id":"102","firstName":"Arlina","lastName":"Bibi","emailBusiness":"[e-mail]"}

Error handling always has to be done, when working with streams. Since we already did the error handling for all streams, because we are using the pipeline method, the sample project is done.

Congratulations. 🚀✨

TL;DR

• The Newline-delimited JSON (ndjson) format works well with streaming data and large sets of data, where each record is processed individually, and it helps to reduce errors.
• Using pipeline simplifies error handling and stream cleanup, and it makes combining streams more readable and maintainable.

Thanks for reading and if you have any questions , use the comment function or send me a message @mariokandut.

If you want to know more about Node, have a look at these Node Tutorials.

References (and Big thanks):

HeyNode,Node.js - Streams,MDN - Streams,Format and MIME Type,ndjson,csvtojson