Question: Parsing CSV-File without starving CPU

[sven42]

Hello,

I am parsing a quite big CSV-File and get a warning from cats-effect about starving my CPU.

Here is the code I use:

import blobstore.url.Url
import cats.effect.IO
import fs2.{text, Pipe, Stream}
import fs2.compression.Compression
import fs2.io.file.{Files, Path}
import fs2.data.csv.*

trait Csv[T]  {

  val decode: Pipe[IO, String, T]
  val encode: Pipe[IO, T, String]

  def streamFrom(path: Path): Stream[IO, T] = decoded(load(path))

  def decoded(stream: Stream[IO, Byte]): Stream[IO, T] = stream.through(text.utf8.decode).through(decode)

  def streamTo(path: Path, data: Stream[IO, T]): IO[Unit] =
    save(path, data.through(encode andThen fs2.text.utf8.encode))

  def load(path: Path): Stream[IO, Byte] = {
    val byteStream = Files[IO].readAll(path)
    if (path.toString.endsWith(".gz")) byteStream.through(gunzip) else byteStream
  }

  val gunzip: Pipe[IO, Byte, Byte] = _.through(Compression[IO].gunzip()).flatMap(_.content)
}

object Csv {
  def apply[T](
      csvSeparator: Char = ';'
  )(implicit rowDecoder: CsvRowDecoder[T, String], rowEncoder: CsvRowEncoder[T, String]): Csv[T] =
    new Csv[T] {
      override val decode: Pipe[IO, String, T] = decodeUsingHeaders[T](csvSeparator)

      override val encode: Pipe[IO, T, String] = encodeUsingFirstHeaders(fullRows = true, csvSeparator)

    }
}

  val historicalDataCsv: Csv[HistoricalData] =
    Csv.apply()(deriveCsvRowDecoder[HistoricalData], deriveCsvRowEncoder[HistoricalData])

  def load(path: Url.Plain): IO[Map[String, HistoricalData]] =
    for {
      -          <- effectLogger.info(s"Loading catalog ${path.path.absolute} ...")
      catalogMap <- historicalDataCsv
                      .streamFrom(path)
                      .map(catalogData => catalogData.key -> catalogData)
                      .compile
                      .toList

      _ <- effectLogger.info(s"Loading catalog succeeded. Found ${catalogMap.size} entries")
    } yield catalogMap.toMap

I print out the thread name when a case class HistoricalData is created and in fact it uses e.g. "io-compute-4"

What can I do to avoid starving CPU?

I already tried to run it with IO.blocking(()).flatMap(_ => ...) without success.

Thanks in advance and best regards,
Sven

[satabin]

Hi Sven, thanks for reporting this issue.
It is expected that the case class is built in the compute pool. Are the individual rows in your file big? If so how big are we talking about?
Running IO.blocking(()).flatMap(_ => ...) cannot help, as the only blocking section in this is the unit part.
Can I also see the definition of save?

If processing individual rows is the part that creates starvation, for now you can try to run the compiled stream in a dedicated execution context with evalOn, but before that, can you share the save method please? Just as a sanity check to be sure I am not looking into the wrong direction.

[sven42]

Hello @satabin ,

thank you for your quick response.

Here is the save method:

  def save(path: Path, data: Stream[IO, Byte]): IO[Unit] = {
    val byteStream = if (path.extName == ".gz") data.through(Compression[IO].gzip()) else data
    byteStream.through(Files[IO].writeAll(path)).compile.drain
  }

The file in dev environment is about 15M with 1.5 Mio lines of quite short lines (because of preprocessing very long lines in raw CVS which was really slow (way slower than doing in in Python/Pandas )).

Processing takes about 4 Seconds; and I expect the file to be way bigger in production.

[satabin]

Interesting, can you try to set a shorter chunk size from the file you read (chunkSize parameter from readAll)? By default the buffer size is 64 * 1024 bytes. Try something like 1024. It might be slower but it might also help with the starvation. I think we need to review the opportunity to insert some cede in our CSV parser chunk processing.

If making the chunk size smaller helps, I think it is worth improving that part in fs2-data.

[sven42]

Hello @satabin,

I changed the chunk size and it got a bit slower but lowering or increasing chunk size does not seem to have an effect concerning starvation.

However, I am not sure, if really fs2-data is causing the starvation. I just saw, that the construction of my case classes is happening with a thread from compute-pool and I wanted to fix it as it seemed to be the remaining work done outside of IO.blocking.

There is quite a lot of things happening in parallel during startup and even scalapy with python using native code is doing heavy stuff.

I will try to test this issue isolated, but currently I am quite busy...

Thanks for your support!