I am converting from Parquet to CSV using javascript.
The example below works, but i am storing in memory the array of values read from Parquet, in records.
Parquet library uses AsyncIterator while the CSV library uses Node Stream API.
I would like to know how to implement a more elegant solution, leveraging streams and reducing memory footprint. TIA
libraries - Parquet: https://github.com/ironSource/parquetjs CSV: https://csv.js.org/
import pts from 'parquets'
let { ParquetSchema, ParquetWriter, ParquetReader } = pts
import * as fs from 'fs'
import stringify from 'csv-stringify'
// declare a schema for the `PI` table
let schema = new ParquetSchema({
Source: { type: 'UTF8' },
TagID: { type: 'UTF8' },
Timestamp: { type: 'TIMESTAMP_MILLIS' },
Value: { type: 'DOUBLE' },
});
const WriterParquet = async () => {
// create new ParquetWriter that writes to 'pi.parquet`
let writer = await ParquetWriter.openFile(schema, 'pi.parquet')
// append a few rows to the file
await writer.appendRow({Source: 'PI/NO-SVG-PISRV01', TagID: 'OGP8TI198Z.PV', Timestamp: new Date(), Value: 410 })
await writer.appendRow({Source: 'PI/NO-SVG-PISRV01', TagID: 'OGP8TI198Z.PV', Timestamp: new Date(), Value: 420 })
await writer.close()
}
const WriterCSV = async () => {
// create new ParquetReader that reads from 'pi.parquet`
let reader = await ParquetReader.openFile('pi.parquet')
// create a new cursor
let cursor = reader.getCursor()
// read all records from the file and print them
let records = []
let record = null;
while (record = await cursor.next()) {
console.log(record)
records.push(record)
}
await reader.close()
// write to CSV
stringify(records, {
header: true
}, function (err, output) {
console.log(output)
fs.writeFile('./pi.csv', output, () => {});
})
}
const Main = async () => {
console.log('writing parquet...')
await WriterParquet()
console.log('reading parquet and writing csv...')
await WriterCSV()
}
Main()
Instead of using the cursor i used the Readable.from(reader) creating a ReadableStream, after this, it was easy to pipe into csv-stringify:
const WriterCSV = async () => {
// create new ParquetReader that reads from 'pi.parquet`
let reader = await ParquetReader.openFile('pi.parquet')
// read all records from the file and print them
const readStream = Readable.from(reader)
readStream.pipe(
stringify({
header: true,
columns: {
Source: 'Source',
TagID: 'TagID',
Timestamp: 'Timestamp',
Value: 'Value'
}
}, function (error, output) {
fs.writeFile('./pi.csv', output, () => {});
}))
readStream.on('end', async function () {
await reader.close();
});
}