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.