Iterating each chunk with async function with stopping stream about scramjet HOT 2 CLOSED

Hi @Peterculazh,

That's indeed how Scramjet works, not only that, but also if you have a number of functions these will run immediately chained when some chunks are resolved.

In order to achieve what you want, you need to inform the framework that you want to run the function one by one like this:

DataStream
  .from(parseStream(stream))
  .setOptions({maxParallel: 1})
  .do(async (row: string[]) => {
      console.log(`Reading row - ${row}`);
      // read row
      // perform async request to another endpoint
      // store in db
      rowNumber++;
  })
  .run()

I would however consider the option to run your ops in a slightly different fashion - split up your workflow and see which methods you can run in parallel, and which you shouldn't - you can gain a lot of performance this way.

DataStream
  .from(parseStream(stream))
  .setOptions({maxParallel: 32})
  .map((row: string[]) => {
    console.log(`Reading row - ${row}`);
    // read and return mapped row
  })
  .map(async (row: MappedRow) => {
    // perform async request to another endpoint
  })
  .setOptions({maxParallel: 1})
  .do(async (row: MappedRowWithExtraData) => {
    // store in db
    rowNumber++;
  })
  .run()

You could potentially use batch() for the DB also.

I hope this solves your issue. Do you think there's a place for a special method in the framework for this? Maybe something like uniDo or uniMap or just uni()?

from scramjet.

A lot thanks. First example is perfectly run as expected and required. As for parallels, I don't need in my case for parallels due of sensitive data, but I got it, thanks.
As for special method, I think uniDo is good

Feel free to close topic, my problem is solved

from scramjet.