Estoy convirtiendo de Parquet a CSV usando javascript.
El siguiente ejemplo funciona, pero estoy almacenando en la memoria la matriz de valores leídos de Parquet, en registros .
La biblioteca Parquet usa AsyncIterator mientras que la biblioteca CSV usa la API Node Stream .
Me gustaría saber cómo implementar una solución más elegante, aprovechando los flujos y reduciendo el consumo de memoria. AIT
bibliotecas - 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()En lugar de usar el cursor , usé Readable.from (lector) creando un ReadableStream , después de esto, fue fácil conectarlo a 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(); }); }