How to partition a large julia DataFrame to an arrow file and process each partition sequentially when reading the data

Question

I am working with very large DataFrames in Julia resulting in out of memory errors when I do joins and other manipulations on the data. Fortunately the the data can be partitioned on an identifier column. I want to persist the partitioned DataFrame using the record batches feature build into Arrow.jl, and then read and process each record batch in turn. I have managed to get the following to work, but are unable to get the original DataFrame back on reading the data. On reading back the data I get a DataFrame with all the data in each column an array of the data in the original partition. I don't know whether my problem is how I am creating the partitions in the first place or on how I am reading back the data:

using Random
using DataFrames
using Arrow

function nextidrange(minId, maxId, batchsize, i)
    fromId = minId + batchsize * (i-1)
    toId = min(maxId, (minId + batchsize * i)-1)
    return fromId, toId
end

minId = 1
maxId = 1000
idrange = (maxId - minId) + 1
df = DataFrame(ID=minId:maxId, B=rand(idrange), C=randstring.(fill(5,idrange)));
batchsize = 100
batches = ceil(Int32, idrange / batchsize)
partitions = Array{SubDataFrame}(undef, 0)
for i = 1:batches
    fromId, toId = nextidrange(minId, maxId, batchsize, i)
    push!(partitions, filter([:ID] => x -> fromId <= x <= toId, df; view = true))
end
io = IOBuffer()
Arrow.write(io, partitions)
seekstart(io)
batches = Arrow.Stream(io)
for b in batches
  bt = b |> DataFrame
  println("Rows = $(nrow(bt))")
end

For each record batch I am expecting a DataFrame with three columns and 100 rows of data. Implementation notes: In the actual data there may be gaps in the identifier values. I have considered using JuliaDB, but DataFrames appears to be much better maintained and supported.

Answer 1

I have resolved my problem, like this:

using Random
using DataFrames
using Arrow
using Tables

function nextidrange(minId, maxId, batchsize, i)
    fromId = minId + batchsize * (i-1)
    toId = min(maxId, (minId + batchsize * i)-1)
    return fromId, toId
end

minId = 1
maxId = 1000
idrange = (maxId - minId) + 1
df = DataFrame(ID=minId:maxId, B=rand(idrange), C=randstring.(fill(5,idrange)));
batchsize = 100
numbatches = ceil(Int32, idrange / batchsize)
partitions = Array{SubDataFrame}(undef, 0)
for i = 1:numbatches 
    fromId, toId = nextidrange(minId, maxId, batchsize, i)
    push!(partitions, filter([:ID] => x -> fromId <= x <= toId, df; view = true))
end
io = IOBuffer()
Arrow.write(io, Tables.partitioner(partitions))
seekstart(io)
recordbatches = Arrow.Stream(io)
ab = Array{DataFrame}(undef,0)
for b in recordbatches 
  bt = b |> DataFrame
  println("Rows = $(nrow(bt))")
  push!(ab,bt)
end

The issue was that the array of DataFrame views should be placed in a call to Tables.partitioner