# Arrow stream usage clarification
**URL:** https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508
**Category:** Data
**Tags:** dataframes, arrow
**Created:** [October 30, 2022, 2:27am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508 "2022-10-30T02:27:09Z")
**Posts on this page:** 11
**Page:** 1
### Author:  [@jeremiedb](https://discourse.julialang.org/u/jeremiedb)
#### Post date: [October 30, 2022, 2:27am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/1 "2022-10-30T02:27:09Z")
I’m trying to use Arrow’s Stream [https://arrow.juliadata.org/dev/reference/#Arrow.Stream](https://arrow.juliadata.org/dev/reference/#Arrow.Stream) to get an efficient out of RAM tabular data iterator.
However, attempts to iterate over the arrow stream only appear to perform a single iteration, where the entire table was read.
For example, here’s the test data creation:
```julia
using DataFrames
using Tables
using Arrow
using Random
using BenchmarkTools
Random.seed!(123)
nobs = 4_000_000
nfeats = 100
path = "data/test.arrow"
x = rand(nobs, nfeats)
df = DataFrame(x, :auto)
Arrow.write(path, df)
```
Now, by itearting over the Stream, the full table get materialized in a DataFrame in a single iteration:
```julia
for x in Arrow.Stream(path)
df = DataFrame(x)
sz = size(df)
println("size: $sz")
end
size: (4000000, 100)
```
My wish regarding Arrow’s Stream was have something behaving like: `itr = Iterators.partition(stream, batch_size)`, so that each iterator would return a table/DataFrame with `batch_size` rows.
The docs read `This allows iterating over extremely large "arrow tables" in chunks represented as record batches.` which seems to point in that direction. I would appreciate to know how to achieve such batch processing.
---
### Author:  [@bkamins](https://discourse.julialang.org/u/bkamins)
#### Post date: [October 30, 2022, 8:43am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/2 "2022-10-30T08:43:21Z")
`Arrow.Stream` reads data in record batches. Therefore you need to write data in record batches to a file first. Here is a simple example of splitting your data into two batches:
```julia
p = Tables.partitioner([view(df, 1:2000000, :), view(df, :2000001:4000000, :)])
Arrow.write(path, p)
```
---
### Author:  [@bkamins](https://discourse.julialang.org/u/bkamins)
#### Post date: [October 30, 2022, 8:52am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/3 "2022-10-30T08:52:16Z")
> [@jeremiedb](#):
>
> to get an efficient out of RAM tabular data iterator.
Also note that DataFrames.jl can work with larger than RAM Arrow.jl files since it is using memory mapping. The only case when you need partitions is when you would need to materialize this data in RAM (which typically, e.g. when doing aggregation, is not needed).
---
### Author:  [@bkamins](https://discourse.julialang.org/u/bkamins)
#### Post date: [October 30, 2022, 9:22am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/4 "2022-10-30T09:22:22Z")
I have added [add Iterators.partition by bkamins · Pull Request #3212 · JuliaData/DataFrames.jl · GitHub](https://github.com/JuliaData/DataFrames.jl/pull/3212) to make partitioning of `DataFrame` objects easier. Until 1.5 release you can just copy the code I used there.
---
### Author:  [@JavierLoaizaRivas](https://discourse.julialang.org/u/JavierLoaizaRivas)
#### Post date: [December 1, 2022, 11:00am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/5 "2022-12-01T11:00:42Z")
This threat is very clarifying, thank so much!.
Nevertheless, there is a case where I still can not see the light: What happens when the source is itself an Arrow.Stream?. In this case it is not obvious to me how to convert each table (in a partion) to a dataframe and then back to a new partition…in my naive understanding something like this should work:
```julia
using Arrow
using DataFrames
using IterTools
MassivePartionedTable_Input = Arrow.Stream("inputFile.arrow")
firstPartition, restPartitions = firstrest( MassivePartionedTable_Input )
df_firstPartition = DataFrame( firstPartition )
largerTableOuput = DoesSomeThingOnThisPartition( df_firstPartition)
Arrow.write("outputFile.arrow",largerTableOutput)
for eachPartition in restPartitions
df_eachPartition = DataFrame( eachPartition)
largerTableOuput = DoesSomeThingOnThisPartition( df_eachPartition )
Arrow.append("outputFile.arrow",largerTableOutput)
end
```
However I obtain as error:
```julia
ERROR: MethodError: no method matching append(::WindowsPath, ::DataFrame)
```
I am not sure about how to proceed…should I convert largerTableOUtput back as table with 1 partition…
sorry…in this case I find the documentation a bit fussy. I really will appreciate your ideas.
Javier
---
### Author:  [@bkamins](https://discourse.julialang.org/u/bkamins)
#### Post date: [December 1, 2022, 11:21am UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/6 "2022-12-01T11:21:15Z")
You need `Arrow.write("outputFile.arrow",largerTableOutput, file=false)` to allow for streaming.
However, this is not needed, you can just use `Arrow.append`:
```julia
MassivePartionedTable_Input = Arrow.Stream("inputFile.arrow")
for eachPartition in MassivePartionedTable_Input
df_eachPartition = DataFrame(eachPartition)
largerTableOuput = DoesSomeThingOnThisPartition( df_eachPartition )
Arrow.append("outputFile.arrow", largerTableOutput)
end
```
Example (input has 2 partitions):
```julia
julia> s = Arrow.Stream("test.arrow")
Arrow.Stream(Arrow.ArrowBlob[Arrow.ArrowBlob(UInt8[0xff, 0xff, 0xff, 0xff, 0xa8, 0x00, 0x00, 0x00, 0x10, 0x00 … 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00], 1, 600)], 1, nothing, Symbol[], Type[], nothing, Dict{Int64, Arrow.DictEncoding}(), Dict{Int64, Arrow.Flatbuf.Field}(), true, Base.RefValue{Union{Nothing, Symbol}}(nothing))
julia> collect(s)
2-element Vector{Any}:
Arrow.Table with 1 rows, 2 columns, and schema:
:a Int64
:b Int64
Arrow.Table with 1 rows, 2 columns, and schema:
:a Int64
:b Int64
julia> for x in s
df = DataFrame(x)
@show df
Arrow.append("test2.arrow", df)
end
df = 1×2 DataFrame
Row │ a b
│ Int64 Int64
─────┼──────────────
1 │ 1 2
df = 1×2 DataFrame
Row │ a b
│ Int64 Int64
─────┼──────────────
1 │ 2 3
julia> read("test.arrow") == read("test2.arrow")
true
```
---
### Author:  [@JavierLoaizaRivas](https://discourse.julialang.org/u/JavierLoaizaRivas)
#### Post date: [December 1, 2022, 12:03pm UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/7 "2022-12-01T12:03:21Z")
Thanks a lot for your super quick reaction.
I did not know that you could also use append in this fashion!. Elegant. There is however a little issue. When I apply this approach to my problem I obtain the following error:
```julia
ERROR: MethodError: no method matching append(::WindowsPath, ::DataFrame)
Closest candidates are:
append(::IO, ::Any; metadata, colmetadata, largelists, denseunions, dictencode, dictencodenested, alignment, maxdepth, ntasks, convert, file) at C:\Users\jloaizar\.julia\packages\Arrow\ZlMFU\src\append.jl:70
append(::IO, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any, ::Any) at C:\Users\jloaizar\.julia\packages\Arrow\ZlMFU\src\append.jl:105
append(::String, ::Any; kwargs...) at C:\Users\jloaizar\.julia\packages\Arrow\ZlMFU\src\append.jl:62
```
May I ask, what version of Arrow and DataFrames are you using?.
Javier
---
### Author:  [@bkamins](https://discourse.julialang.org/u/bkamins)
#### Post date: [December 1, 2022, 2:52pm UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/8 "2022-12-01T14:52:05Z")
```julia
(@v1.8) pkg> st Arrow
[69666777] Arrow v2.4.1
(@v1.8) pkg> st DataFrames
[a93c6f00] DataFrames v1.4.3
```
But the problem seems the `::WindowsPath` part. Are you sure you are passing a string with a file name? Can you also pass full stack trace?
---
### Author:  [@JavierLoaizaRivas](https://discourse.julialang.org/u/JavierLoaizaRivas)
#### Post date: [December 1, 2022, 4:14pm UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/9 "2022-12-01T16:14:59Z")
That was indeed the problem. In my code I used FilePaths…and the resulting object was interpreted correctly by Arrow.write, and unfortunately Arrow.merge did not know what to do with it. After a few changes it all works well now.
Below the complete solution, as reference for someone experiencing the same problem:
```julia
using Arrow
using DataFrames
using IterTools
using FilePaths
OutputDir = absolute(p"outputDirectory")
OutputFile="outputFile.arrow"
OutputPath = joinpath(OutputDir,OutputFile);
MassivePartionedTable_Input = Arrow.Stream("inputFile.arrow")
firstPartition, restPartitions = firstrest( MassivePartionedTable_Input )
df_firstPartition = DataFrame( firstPartition )
largerTableOuput = DoesSomeThingOnThisPartition( df_firstPartition)
Arrow.write(OutputPath,largerTableOutput,file=false)
for eachPartition in restPartitions
df_eachPartition = DataFrame( eachPartition)
largerTableOuput = DoesSomeThingOnThisPartition( df_eachPartition )
Arrow.append(string(OutputPath),largerTableOutput)
end
```
Looking inside the Arrow sources I found something caller Arrow.writer. Unfortunately I could not find any example on how to use it. Do you know what would be the solution if one were to use Arrow.Writer rather than the combination of Arrow.write and Arrow.append that was used in this case?
Thanks again for all your help, I learned a lot today.
Sincerely
Javier
Javier
---
### Author:  [@bkamins](https://discourse.julialang.org/u/bkamins)
#### Post date: [December 1, 2022, 5:13pm UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/10 "2022-12-01T17:13:41Z")
AFACIT `Writer` is an internal name only and should not be used by users.
(as a side note as commented above: you do not need to use `Arrow.write` in your solution, using `Arrow.append` should be enough)
---
### Author:  [@Jake](https://discourse.julialang.org/u/Jake)
#### Post date: [July 17, 2023, 7:55pm UTC](https://discourse.julialang.org/t/arrow-stream-usage-clarification/89508/11 "2023-07-17T19:55:57Z")
> [@bkamins](#):
>
> AFACIT `Writer` is an internal name only and should not be used by users.
This appears to be old information. Looking at the [docs](https://arrow.apache.org/julia/dev/reference/#Arrow.Writer) shows support and examples.