NodeJS data streaming api CSV and JSON example

Node.js provides a powerful streaming API that allows you to efficiently process data in chunks, rather than loading entire files or data sets into memory. This is particularly useful for handling large files, network protocols, or any data that can be processed incrementally. Let's explore Node.js streaming API with an example:

Example: Reading a Large File and Transforming Data

In this example, we'll create a Node.js script that reads a large CSV file, processes each row, and writes the transformed data to a new file using streams. This will demonstrate both readable and writable streams in action.

Step 1: Setup

First, make sure you have Node.js installed on your system. Create a new directory for your project and initialize a new Node.js project:

mkdir stream-example
cd stream-example
npm init -y

Step 2: Install Required Packages

For this example, we'll use csv-parser to parse CSV files and through2 for creating a transform stream. Install these packages using npm:

npm install csv-parser through2

Step 3: Example Script

Create a file named stream-example.js and add the following code:

const fs = require('fs');
const csv = require('csv-parser');
const through2 = require('through2');

// Input and output file paths
const inputFile = 'input.csv';
const outputFile = 'output.json';

// Create a readable stream to read the CSV file
const readStream = fs.createReadStream(inputFile, 'utf8');

// Create a writable stream to write JSON data
const writeStream = fs.createWriteStream(outputFile, { encoding: 'utf8' });

// Use csv-parser to parse each row of the CSV
readStream
  .pipe(csv())
  // Transform each row and convert to JSON format
  .pipe(through2.obj(function (data, encoding, callback) {
    // Transform the data here (example: converting CSV to JSON)
    const transformedData = {
      id: data.id,
      name: data.name,
      email: data.email.toLowerCase(), // Example transformation: convert email to lowercase
    };
    
    // Pass transformed data to the next stream
    this.push(JSON.stringify(transformedData) + '\n');
    callback();
  }))
  // Write the transformed data to the output file
  .pipe(writeStream)
  .on('finish', () => {
    console.log('Data transformation completed.');
  })
  .on('error', (err) => {
    console.error('Error:', err);
  });

Explanation:

1. Require Modules: We require fs (File System), csv-parser, and through2 modules. csv-parser is used to parse CSV data, and through2 is used to create a transform stream.

2. File Paths: Define inputFile as the path to your CSV file and outputFile as the path where transformed data will be written.

3. Streams Creation:

   • fs.createReadStream(inputFile, 'utf8'): Creates a readable stream to read data from input.csv.
   • fs.createWriteStream(outputFile, { encoding: 'utf8' }): Creates a writable stream to write transformed data to output.json.

4. Piping Streams:

   • .pipe(csv()): Pipes the readable stream (readStream) through csv-parser to parse each row of the CSV file into a JavaScript object.
   • .pipe(through2.obj(...)): Pipes each parsed row through a transform stream created using through2. Here, through2.obj creates a transform stream where each row is transformed into JSON format (transformedData).

5. Transform Function (through2.obj):

   • through2.obj(function (data, encoding, callback) { ... }): This function is called for each row of data. Here, we transform the data (in this example, converting email to lowercase) and push the transformed JSON string (JSON.stringify(transformedData)) to the writable stream.

6. Error and Completion Handling:

   • .on('finish', () => { ... }): Logs a message when the data transformation is completed.
   • .on('error', (err) => { ... }): Logs an error message if there's an error during the stream operations.

Step 4: Run the Script

Save the changes to stream-example.js and run the script using Node.js:

node stream-example.js

This will read input.csv, transform each row of data according to the specified transformation (converting email to lowercase in this example), and write the transformed data to output.json using Node.js streaming API.