Write CSV row by row

Specific Domains Data
,
1

I have a computation that calculates rows of data, let’s say they are NamedTuples with the same keys. I would like to emit this into a CSV file without collecting or making it an iterable.

Why?

Why not collect? The data is large. Why not wrap in an iterable? The algorithm is such that it is easier to pass around a sink to which one writes to, than concatenate many layers of iterators.

Pseudocode for what I want:

sink = make_csv_sink(path, schema::Type{<:NamedTuple{colnames}})

emit(sink, a_namedtuple) # checking that colnames match would be nice

close(sink)

When the fields are numbers or unescaped strings, this can be trivially implemented, but if we get into fancier CSV escapes it becomes tricky. This question has been asked before without a solution.

Existing CSV libraries must have this functionality (or the building blocks for it), but it is not exposed.

2

This would be a useful feature. Writing rows directly to a CSV without collecting everything first feels like a common use case, especially for long running computations. It would be nice if the API also validated the schema on each write.

3

There’s CSV.writerow, which is not public unfortunately but seems to be closest to what you’re looking for.

4

Maybe we can make this part more straightforward in Julia? Turning such a “sink-able” algorithm into an iterable. Seems like that’s the right seam to plug in, and useful way beyond CSV writing.

5

Do you have a suggestion on how to approach this? In my mind they are two fundamentally opposed approaches.

One way I could imagine is a Channel, which blocks iterate unless it has stuff in the queue, which can then be closed and then it would return nothing. But it is somewhat convoluted.

6

Sounds similar to what Python’s yield / yield from solve… Your Channel suggestion doesn’t seem wild as the base for an implementation in Julia. Turning callback-based algorithm into an iterator would enable things like “save only each 10th entry to CSV” or filter etc, and probably the cleaner way fundamentally.

7

I’m unclear about the restriction about not “making it an iterable”. Why not use eachrow?

julia> using DataFrames, CSV

julia> df = DataFrame(a=[1,2,3,4,5], b=["Tamas_Papp", "NathanEvans", "langestefan", "aplavin", "mkitti"])
5×2 DataFrame
 Row │ a      b           
     │ Int64  String      
─────┼────────────────────
   1 │     1  Tamas_Papp
   2 │     2  NathanEvans
   3 │     3  langestefan
   4 │     4  aplavin
   5 │     5  mkitti

julia> open("cool.csv", "w") do f
           for row in eachrow(df)
               CSV.write(f, DataFrame(row), append=true)
           end
       end

julia> read("cool.csv", String) |> println
1,Tamas_Papp
2,NathanEvans
3,langestefan
4,aplavin
5,mkitti

Here’s a variation that is perhaps closer to your original prompt.

julia> using DataFrames, CSV

julia> nts = @NamedTuple{a::Int64, b::String}[
           (1,"Tamas_Papp"),
           (2,"NathanEvans"),
           (3,"langestefan"),
           (4,"aplavin"),
           (5,"mkitti")
       ]
5-element Vector{@NamedTuple{a::Int64, b::String}}:
 (a = 1, b = "Tamas_Papp")
 (a = 2, b = "NathanEvans")
 (a = 3, b = "langestefan")
 (a = 4, b = "aplavin")
 (a = 5, b = "mkitti")

julia> stateful = Iterators.Stateful(nts)
Base.Iterators.Stateful{Vector{@NamedTuple{a::Int64, b::String}}, Union{Nothing, Tuple{@NamedTuple{a::Int64, b::String}, Int64}}}([(a = 1, b = "Tamas_Papp"), (a = 2, b = "NathanEvans"), (a = 3, b = "langestefan"), (a = 4, b = "aplavin"), (a = 5, b = "mkitti")], ((a = 1, b = "Tamas_Papp"), 2))

julia> f = open("cool.csv", "w")
IOStream(<file cool.csv>)

julia> CSV.write(f, DataFrame([first(iterate(stateful))]), append=true, header=true)
IOStream(<file cool.csv>)

julia> CSV.write(f, DataFrame([first(iterate(stateful))]), append=true)
IOStream(<file cool.csv>)

julia> CSV.write(f, DataFrame([first(iterate(stateful))]), append=true)
IOStream(<file cool.csv>)

julia> CSV.write(f, DataFrame([first(iterate(stateful))]), append=true)
IOStream(<file cool.csv>)

julia> CSV.write(f, DataFrame([first(iterate(stateful))]), append=true)
IOStream(<file cool.csv>)

julia> close(f)

julia> read("cool.csv", String) |> println
a,b
1,Tamas_Papp
2,NathanEvans
3,langestefan
4,aplavin
5,mkitti
8

Because that requires collecting all the available data beforehand. And at that point, you might as well write them all at once to the sink. The emit method can work async, you write a new line only when you read new data.

9

CSV.jl 1.0 still does not expose a persistent emit writer. But this gives your computation a push-style callback using only the public API:

using CSV

R = @NamedTuple{id::Int, label::String}

open("out.csv", "w") do io
    CSV.write(io, R[])  # Write the header from the empty typed table.
    emit(row::R) = CSV.write(io, (row,); writeheader=false)

    emit((id=1, label="with,comma"))
    emit((id=2, label="a \"quote\"\nand a newline"))
    # Pass emit into the rest of your computation.
end

This keeps the file open and delegates escaping to CSV.jl. The callback requires the declared field names, order, and types. No previous rows are collected. Each call wraps just its own row in a one-element tuple.

It does repeat writer setup for each row. I’d be open to a PR to publicly expose something like this if anyone’s interested.