CSV Component

[skhatri]

I can write CSV component, if you like.

I was thinking of creating a CSVFlow as well as CSVSourceStage.
Where CSVFlow will take a ByteString and chunk it, the CSVSourceStage could be a wrapper around FileIO. Let me know if this will be an acceptable component.

[hseeberger]

AFAIK Camel has it, too, so this makes sense to me.

[stephennancekivell]

Great idea!
Might be nice to wrap open csv or something to take care of the escaping nuance's.

Or scala-csv / pure-csv / mighty csv.

[joearasin]

Not sure if it's optimal, but we've been using something along the lines of this to write CSVs, built around commons-csv:

  def forFormat(csvFormat: CSVFormat, charset: Charset = StandardCharsets.UTF_8): Flow[Seq[Any], ByteString, NotUsed] = {
    Flow[Seq[Any]].map {
      cells => {
        val buf: ByteStringBuilder = ByteString.newBuilder
        val writer = new OutputStreamWriter(buf.asOutputStream, charset)
        csvFormat.printRecord(writer, cells.map(_.asInstanceOf[AnyRef]): _*)
        writer.close()
        buf.result()
      }
    }
  }

On top of this, one thing worth keeping ind mind with an API sort of like this is possibly padding the inputs, to deal with varying-length sequences that may be truncated.

[ennru]

@skhatri Did you put any more effort into this?

What I find in the existing CSV libraries (commons-csv, open csv, scala-csv) is that they tend to take control over file IO, which is not what we want in an Akka Stream. I've done some parsing Stages with scala-csv, but converting to CSV is a bit like what @joearasin shows for commons-csv.

[johanandren]

On the parser side we'd want it to be push/record based, in that we can push bytestrings on it and potentially get a record back, just like the JSON framing in Akka stream core: https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/scaladsl/JsonFraming.scala

[ennru]

So the expected signature could be something like
val csvFlow: Flow[ByteString, Source[ByteString, NotUsed], NotUsed] = CsvFraming()
where the created stream contains one element per column in the incoming CSV ByteString and a new element for every "line" (CSV columns may contain line breaks).

Or should it make things simpler and create elements of immutable.Seq[ByteString].

[johanandren]

Or the stage would both frame and parse and emit CsvRecords or something like that?

[johanandren]

Can you think of a use-case where it is interesting to consume fields rather than entire records?

[ennru]

That would be this extra-wide Excel data a former colleague of mine produced, in case I'd only be interested in the first 100 columns.
But the framing would need to go through the rest anyway, so a fixed length thingy will be best. I would not want to introduce a CsvRecord. Another step in the stream would be to convert the list of elements to a map out of the column names (Map[String, String]), which on request could be read as the first record in the stream.

[johanandren]

Ok, sounds like you have a plan! 👍