To convert JSON to CSV, I love using json2csv. It really does all the hard work of working the JSON structure out and converting it to a flat file. For nested JSON elements, you can simply specify them by the dot notation (like transaction.itemId). When it contains an array element, it automatically expands it and creates a denormalised table for you. Well, let’s see how it works.

Input Data

Here is the input data, transaction.json. I think it is a good representation of standard JSON file you are likely to ingest. It has nested elements in Payment and an array element in Transaction. It follows schemaless patterns with missing elements when they do not exist in the dataset.

Setup

As usual, let’s create a project folder and install required modules.

We also need to create a table that contains all the elements in the source file.

Code

For bulk loading data in to Postgres, see the post here. The key to convert JSON to CSV is to define the fields and which element to unwind. As Transaction has an array, we will unwind with it. This will automatically create a denormalised table.

There are a few different ways to convert json to csv as in the documentation. In this example, we are going to use readable stream and pipe the conversion function. This method is memory efficient especially when you have a large JSON file.

After json2csv emits the ‘end’ event, we will call the query execution function. This will ensure that the queries get executed after the csv file creation.

// Loading required modules
const fs = require('fs')
const path = require('path')
const Json2csvTransform = require('json2csv').Transform
const {Client} = require('pg')
const copyFrom = require('pg-copy-streams').from
const config = require('./config.json')

// Defining file path
const inputFile = path.join(__dirname, '/data/transaction.json')
const outputFile = path.join(__dirname, '/data/transaction.csv')

// Defining fields
const fields = ['Id', 'Name', 'TransactionId', 'Transaction.ItemId',
'Transaction.price', 'Subscriber', 'Payment.Type', 'Payment.Total',
 'Payment.Success', 'Note']

// target table
var table = 'usermanaged.trans'

// Getting connectin parameters from config.json
const host = config.host
const user = config.user
const pw = config.pw
const db = config.db
const port = config.port
const conString = `postgres://${user}:${pw}@${host}:${port}/${db}`

const executeQuery = (targetTable) => {
    console.log('Starting executeQuery function')
    // Connecting to Database
    const client = new Client({
        connectionString: conString,
    })
    client.connect()

    const execute = (target, callback) => {
        client.query(`Truncate ${target}`, (err) => {
                if (err) {
                client.end()
                callback(err)
                // return console.log(err.stack)
                } else {
                console.log(`Truncated ${target}`)
                callback(null, target)
                }
            })
    }
    execute(targetTable, (err) =>{
        if (err) return console.log(`Error in Truncate Table: ${err}`)
        var stream = client.query(copyFrom(`COPY ${targetTable} FROM STDIN CSV HEADER`))
        var fileStream = fs.createReadStream(outputFile)
        
        fileStream.on('error', (error) =>{
            console.log(`Error in creating read stream ${error}`)
        })
        stream.on('error', (error) => {
            console.log(`Error in creating stream ${error}`)
        })
        stream.on('end', () => {
            console.log(`Completed loading data into ${targetTable}`)
            client.end()
        })
        fileStream.pipe(stream);
    })
}

// Create a readable stream
const input = fs.createReadStream(inputFile, { encoding: 'utf8' });
const output = fs.createWriteStream(outputFile, { encoding: 'utf8' });
const json2csv = new Json2csvTransform({fields, unwind:'Transaction'});
 
const processor = input.pipe(json2csv).pipe(output);
 
// You can also listen for events on the conversion and see how the header or the lines are coming out.
json2csv
  .on('header', header => console.log(header))
  .on('line', line => console.log(line))
  .on('error', err => console.log(err))
  .on('end', () =>{
      console.log(`Writing csv file, ${outputFile}, has been completed`)
      // After csv file is created, execute the query to load the data
      executeQuery(table)
  }) 

Fun!