Read csv files from zip file

[wrnrlr]

I am trying to the files inside a zip file using a transform stream all wrapped into a iterator for convenience.

I haven't gotten to the csv part yet because I can't figure out how to use the FileEntry.getData with another stream reader.
My current implementation also has the problem is that is hangs for a while after having processed the zip file. This is obvious from looking at the log where the last stream is aborted after END.

Another problem is that the iterator only yields one line if I put a TextLineStream in between.

From reading the doc I suspect I might need to have access to the defaultWriter to ease out more values and close it after processing, but the implementation of getData is really hard to follow.

import {Uint8ArrayWriter,Uint8ArrayReader,TextReader,ZipReader,ZipWriter} from 'https://deno.land/x/zipjs/index.js'
import {TextLineStream} from 'https://deno.land/std/streams/mod.ts'
import {readCSV,CSVReader} from 'https://deno.land/x/csv/mod.ts'

class FileIterator {
  accumulator = []
  constructor(entry) {
    const transform = new TransformStream()
    const queuingStrategy = new CountQueuingStrategy({highWaterMark:1})
    let accumulator = this.accumulator
    this.writable = new WritableStream({
      write(chunk) {console.log('line');accumulator.push(chunk)},
      close() {console.log('abort csv file iterator')},
      abort(e) {console.error('abort csv file iterator: ',e)}
    }, queuingStrategy)
    transform.readable
      .pipeThrough(new TextDecoderStream())
      // .pipeThrough(new TextLineStream())
      .pipeTo(this.writable)
    this.transform = transform
    this.entry = entry
  }
  [Symbol.asyncIterator]() {return this.iterateFromOffset(0)}
  async *iterateFromOffset(offset) {
    const writer = await this.entry.getData(this.transform)
    while(this.accumulator[offset]) {
      yield this.accumulator[offset++]
    }
  }
}

const zipWriter = new ZipWriter(new Uint8ArrayWriter());
await Promise.all([
  zipWriter.add("a.csv", new TextReader("a,b,c\n1,2,3")),
  zipWriter.add("b.csv", new TextReader("d,e,f\n4,5,6"))])
const zipFile = await zipWriter.close()
const zipReader = new ZipReader(new Uint8ArrayReader(zipFile))
for (const entry of await zipReader.getEntries()) {
  console.log('file ', entry.filename)
  const iter = new FileIterator(entry)
  for await (const content of iter) {
    console.log(content)
  }
}
await zipReader.close()
console.log('END')

log:

file  a.csv
line
a,b,c
1,2,3
file  b.csv
abort csv file iterator
line
d,e,f
4,5,6
END
abort csv file iterator

[wrnrlr]

How do you consume the stream of a file entity as another stream without it leaking a resource or hanging?

[gildas-lormeau]

Sorry again, I think I just realized the problem. You have to call zip.terminateWorkers() once you don't need zip.js anymore (i.e. after await zipReader.close()) otherwise the web workers timeout prevents the program from stopping immediately.

BTW, the log message is incorrect in the WritableStream#close method, it should be "close csv file iterator".

[wrnrlr]

Yes, well the issue of the runtime hanging at the end of execution is resolved.
I am still struggling to turn the stream from the zip file into a csv-row iterator:

I don't quite understand how to "yield" values from inside a callback.
I suspect that at some point I have to make a queue with two different kind of task, one to "feed" the csv parser and another to yield from the parsed csv results, and loop over this queue until it is empty.

But this is probably not really a zip.js issue, but if you would have any tips I would appreciate that.

import {ZipReader,ZipWriter,terminateWorkers,Uint8ArrayWriter,Uint8ArrayReader,TextReader} from 'https://deno.land/x/zipjs@v2.7.29/index.js'
import {initParser,inferSchema} from 'https://esm.sh/udsv@0.5.3'

// https://stackoverflow.com/questions/50862698/how-to-convert-node-js-async-streaming-callback-into-an-async-generator
async function csvFromZipEntry(entry) {
  let csvParser
  const csvReducer = function (rows) {console.log('reduce #rows', rows.length)}
  const write = chunk => {
    console.log('write chunk with length: ', chunk.length)
    csvParser ??= initParser(inferSchema(chunk));
    csvParser.chunk(chunk, csvParser.typedArrs, csvReducer);
  }
  const transform =  new TransformStream()
  transform.readable
    .pipeThrough(new TextDecoderStream())
    .pipeTo(new WritableStream({
      write,
    }, new CountQueuingStrategy({highWaterMark:1}))).then(()=>{console.log('stream has ended')})
  await entry.getData(transform,{}).then(()=>{console.log('getData finished')})
  csvParser.end()
  let done = false
  return {
    [Symbol.asyncIterator]() {return this},
    next(x) {
      if (done) return Promise.resolve({done:true})
      done = true
      // DUNO what to do now...
      return {value:"hello"}
    },
    return(value) {console.log('return');return{value,done:true}},
    throw(e) {console.log('error',e);return {done:true}}
  }
}

const zipWriter = new ZipWriter(new Uint8ArrayWriter());
await Promise.all([
  zipWriter.add("a.csv", new TextReader("a,b,c\n"+Array.from({length:600_000}).map((_,i)=>`${i},${i},${i}`).join('\n'))),
  zipWriter.add("b.csv", new TextReader("d,e,f\n4,5,6"))])
const zipFile = await zipWriter.close()
const zipReader = new ZipReader(new Uint8ArrayReader(zipFile))
for (const entry of await zipReader.getEntries()) {
  console.log('file ', entry.filename)
  for await (const content of await csvFromZipEntry(entry)) {
    console.log('content ===>',content)
  }
}
await zipReader.close()
terminateWorkers()

[gildas-lormeau]

csvFromZipEntry could return an iterable based on ReadableStream which yields each row, see the example below and https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream#async_iteration for more info.

import {
  ZipReader,
  ZipWriter,
  terminateWorkers,
  Uint8ArrayWriter,
  Uint8ArrayReader,
  TextReader
} from 'https://deno.land/x/zipjs@v2.7.29/index.js';
import {
  initParser,
  inferSchema
} from 'https://esm.sh/udsv@0.5.3';

const zipWriter = new ZipWriter(new Uint8ArrayWriter());
await Promise.all([
  zipWriter.add('a.csv', new TextReader(
    'a,b,c\n' + Array.from({ length: 100 }).map((_, i) => `${i},${i},${i}`).join('\n')
  )),
  zipWriter.add('b.csv', new TextReader('d,e,f\n4,5,6'))]);
const zipFile = await zipWriter.close();
const zipReader = new ZipReader(new Uint8ArrayReader(zipFile));
for (const entry of await zipReader.getEntries()) {
  console.log('file ', entry.filename);
  for await (const row of await csvFromZipEntry(entry)) {
    console.log('row =>', row);
  }
}
await zipReader.close();
terminateWorkers();

async function csvFromZipEntry(entry) {
  let csvParser;
  const { readable, writable } = new TransformStream();
  const readableOutput = readable
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new TransformStream({
      transform(chunk, controller) {
        csvParser ??= initParser(inferSchema(chunk));
        csvParser.chunk(chunk, csvParser.typedArrs, rows => rows.forEach(row => controller.enqueue(row)));
      },
      flush() {
        csvParser.end();
      }
    }))
  await entry.getData(writable);
  return readableOutput;
}

[wrnrlr]

I'de love to but I can't seem to find the button where I can do this

[gildas-lormeau]

No problem, it's okay, I'll leave the discussion open otherwise it's no longer visible.

[leeoniya]

probably make sure to pull out the csvReducer fn def out of the chunk loop, so it's not re-creating the arrow fn on each 64kb chunk :)

[gildas-lormeau]

I don't think it's possible since it depends on controller. However, the performance could be improved by not using await when calling getData and csvFromZipEntry. The exception getData might trow can also be ignored, because the stream (i.e. readable) can propagate it.

Here's below the code with these changes applied.

import {
  ZipReader,
  ZipWriter,
  terminateWorkers,
  Uint8ArrayWriter,
  Uint8ArrayReader,
  TextReader
} from 'https://deno.land/x/zipjs@v2.7.29/index.js';
import {
  initParser,
  inferSchema
} from 'https://esm.sh/udsv@0.5.3';

class CSVDecoderStream extends TransformStream {
  constructor() {
    super();
    let csvParser;
    const csvReducer = (controller, rows) => rows.forEach(row => controller.enqueue(row));
    const readable = super.readable
      .pipeThrough(new TextDecoderStream())
      .pipeThrough(new TransformStream({
        transform(chunk, controller) {
          csvParser ??= initParser(inferSchema(chunk));
          csvParser.chunk(chunk, csvParser.typedArrs, rows => csvReducer(controller, rows));
        },
        flush() {
          csvParser.end();
        }
      }));
    Object.defineProperty(this, 'readable', {
      get() {
        return readable;
      }
    });
  }
}

function csvFromZipEntry(entry) {
  return new Promise((resolve, reject) => {
    const { readable, writable } = new CSVDecoderStream();
    entry.getData(writable).catch(reject);
    resolve(readable);
  });
}

async function test() {
  const zipWriter = new ZipWriter(new Uint8ArrayWriter());
  await Promise.all([
    zipWriter.add('a.csv', new TextReader(
      'a,b,c\n' + Array.from({ length: 100 }).map((_, i) => `${i},${i},${i}`).join('\n')
    )),
    zipWriter.add('b.csv', new TextReader('d,e,f\n4,5,6'))]);
  const zipFile = await zipWriter.close();
  const zipReader = new ZipReader(new Uint8ArrayReader(zipFile));
  for (const entry of await zipReader.getEntries()) {
    console.log('file ', entry.filename);
    for await (const row of await csvFromZipEntry(entry)) {
      console.log('row =>', row);
    }
  }
  await zipReader.close();
  terminateWorkers();
}

test().catch(console.error);