# How to subset a stream I/O and pass it to TensorFlow.jl? (question has been updated)

**URL:** <https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502>\
**Category:** Data\
**Tags:** question\
**Created:** [May 3, 2017, 4:30pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502 "2017-05-03T16:30:48Z")\
**Posts on this page:** 20\
**Page:** 2

<div class="post-metadata">

**Author:** ![js135005](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/js135005/32/8219_2.png) [@js135005](https://discourse.julialang.org/u/js135005)\
**Post date:** [May 4, 2017, 1:16am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/22 "2017-05-04T01:16:31Z")

</div>

Basically the DataStream object behaves just like any regular file. If you would read from CSV files, then you can also read directly from the gzipped versions just as well. In my work, I am dealing with hundreds and thousands of multi-gigabyte log files. It is a major advantage to be able to process the compressed files directly. I’m also working with historic data - results from very large scale computer performance load tests.

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 1:26am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/23 "2017-05-04T01:26:35Z")

</div>

Okay so I will work towards reading from GZ and writing in CSV for TensorFlow.jl, unless Tensorflow can also read GZ, which apparently it [can](https://www.tensorflow.org/extend/new_data_formats). I would have to create a new reader inside TensorFlow.jl. @malmaud

I don’t know if it makes a difference to pass the IO stream to TensorFlow.jl as CSV or GZ.

David, how would passing an IO stream to `CSV.source` look like?

---

<div class="post-metadata">

**Author:** ![davidanthoff](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/davidanthoff/32/223493_2.png) [@davidanthoff](https://discourse.julialang.org/u/davidanthoff)\
**Post date:** [May 4, 2017, 2:40am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/24 "2017-05-04T02:40:20Z")

</div>

I believe you are using Query.jl version 0.3.x, but all my examples require version 0.4.0. A simple `Pkg.update()` should get you the latest version of Query and all other packages you are using.

> [@hpoit](#):
>
> David, how would passing an IO stream to CSV.source look like?

I believe you just pass it to the `CSV.Source` constructor like this, assuming you assigned the stream to a variable called `stream`:

```julia
q = @from i in CSV.Source(stream) begin
    @where i.Type .== "Trade"
    @select {i.Price, i.Volumne}
end

```

Just to clarify, when we are talking about a `stream` here that you might get from some unzip package, we are not talking about some streaming data model, we are just talking about a julia IO stream, see the doc here [https://docs.julialang.org/en/latest/manual/networking-and-streams.html](https://docs.julialang.org/en/latest/manual/networking-and-streams.html) about details.

---

<div class="post-metadata">

**Author:** ![Tamas\_Papp](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/tamas_papp/32/25949_2.png) [@Tamas\_Papp](https://discourse.julialang.org/u/Tamas_Papp)\
**Post date:** [May 4, 2017, 5:56am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/25 "2017-05-04T05:56:30Z")

</div>

Depends, and you should benchmark it for your case. I am using the excellent `BioJulia/Libz.jl`, and my bottleneck was IO, so I got a speedup when I was reading data from a HDD and a network drive (with a very fast connection), and got no noticable slowdown on an SSD. So for me it was a no-brainer to use compressed files.

---

<div class="post-metadata">

**Author:** ![joshbode](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/joshbode/32/260_2.png) [@joshbode](https://discourse.julialang.org/u/joshbode)\
**Post date:** [May 4, 2017, 6:19am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/26 "2017-05-04T06:19:08Z")

</div>

Great thread!

Looks like it might need more work with a Libz (BufferedStreams) stream, as CSV.jl requires the `nb_available` method:

```julia
julia> using CSV, Query, Libz

julia> io = Libz.ZlibInflateInputStream(open("/tmp/abc.txt.gz", "r"))
BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}}(<8.0 KiB buffer, 0% filled>)

julia> q = @from i in CSV.Source(io) begin
           @where i.n >= 500
           @select {i.n}
       end
ERROR: MethodError: no method matching nb_available(::BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}})
Closest candidates are:
  nb_available(::Base.Filesystem.File) at filesystem.jl:162
  nb_available(::BufferStream) at stream.jl:1036
  nb_available(::IOStream) at iostream.jl:146
  ...
 in #Source#12(::BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}}, ::CSV.Options, ::Int64, ::Int64, ::Array{DataType,1}, ::Bool, ::Bool, ::Int64, ::Int64, ::Int64, ::Bool, ::Type{T}) at /Users/josh/.julia/v0.5/CSV/src/Source.jl:59
 in (::Core.#kw#Type)(::Array{Any,1}, ::Type{CSV.Source}) at ./<missing>:0
 in #Source#11(::UInt8, ::UInt8, ::UInt8, ::String, ::Int64, ::Int64, ::Array{DataType,1}, ::Bool, ::Bool, ::Base.Dates.DateFormat, ::Int64, ::Int64, ::Int64, ::Bool, ::Type{T}, ::BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}}) at /Users/josh/.julia/v0.5/CSV/src/Source.jl:25
 in CSV.Source(::BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}}) at /Users/josh/.julia/v0.5/CSV/src/Source.jl:21

```

Where `/tmp/abc.txt` (unzipped) is

```julia
n
1
2
...
999
1000

```

Working on the uncompressed stream does work, though 🙂

```julia
julia> q = @from i in CSV.Source(open("/tmp/abc.txt", "r")) begin
           @where i.n >= 500
           @select {i.n}
       end

julia> collect(q)
501-element Array{NamedTuples._NT_n{Nullable{Int64}},1}:
 (n = 500)
 (n = 501)
...
 (n = 999)
 (n = 1000)

```

---

<div class="post-metadata">

**Author:** ![joshbode](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/joshbode/32/260_2.png) [@joshbode](https://discourse.julialang.org/u/joshbode)\
**Post date:** [May 4, 2017, 6:36am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/27 "2017-05-04T06:36:09Z")

</div>

Quick hack to get Zlib streaming working:

```julia
julia> using CSV, Query, Libz

julia> Base.nb_available{T <: Libz.Source}(s::BufferedStreams.BufferedInputStream{T}) = eof(s) # the hack!

julia> io = Libz.ZlibInflateInputStream(open("/tmp/abc.txt.gz", "r"))

julia> q = @from i in CSV.Source(io) begin
           @where i.n >= 500
           @select {i.n}
       end

julia> collect(q)
501-element Array{NamedTuples._NT_n{Nullable{Int64}},1}:
 (n = 500)
 (n = 501)
...
 (n = 999)
 (n = 1000)

```

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 11:10am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/28 "2017-05-04T11:10:02Z")

</div>

Hi David, the latest [results](https://gist.github.com/hpoit/6b26f41a12fc544fa97bb6d8c19f8c5a) 🙂.

I will break fast and return to all your comments below the first two sentences, and to Tamas’ and Josh’s comments.

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 3:32pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/29 "2017-05-04T15:32:23Z")

</div>

Hi David, Tamas, and Josh, thank you all for the information. What I’m set to do then is

```julia

using CSV, DataFrames, Query, Libz

# nb_available hack for streams (Josh correct me if I'm wrong about the reason for the hack)
Base.nb_available{T <: Libz.Source}(s::BufferedStreams.BufferedInputStream{T}) = eof(s)

# Define stream I/O
io = Libz.ZlibInflateInputStream(open("/file.csv.gz", "r"))

# Define query or subset in I/O
q = @from i in CSV.Source(io) begin
    @select i.Time_G_
    @where i.Type == "Trade"
    @select {i.Price, i.Volume}
end

# Define df and build DataFrame from I/O
df = DataFrame(take(q, 10_000_000))

```

I just need a little help from David to get `@select i.Time_G_` right to have the column `Time_G_` displayed next to columns Price and Volume.

Also, Josh, how long do you think this hack will last for? How do I build an insurance policy for it?

---

<div class="post-metadata">

**Author:** ![davidanthoff](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/davidanthoff/32/223493_2.png) [@davidanthoff](https://discourse.julialang.org/u/davidanthoff)\
**Post date:** [May 4, 2017, 4:18pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/30 "2017-05-04T16:18:51Z")

</div>

> [@hpoit](#):
>
> I just need a little help from David to get @select i.Time\_G\_ right to have the column Time\_G\_ displayed next to columns Price and Volume.

You want just one `@select` statement in your query that picks all the columns you need:

```julia
q = @from i in CSV.Source(io) begin
    @where i.Type == "Trade"
    @select {i.Price, i.Volume, i.Time_G_}
end

```

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 4:56pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/31 "2017-05-04T16:56:00Z")

</div>

> [@Tamas\_Papp](#):
>
> Depends, and you should benchmark it for your case.

where should I place `@time`?

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 5:03pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/32 "2017-05-04T17:03:51Z")

</div>

I had tried `@select {i.Time_G_, i.Price, i.Volume}` and that didn’t work so I didn’t think of `@select {i.Price, i.Volume, i.Time_G_}`. So the rule is place columns that apply to `@where` first in the list, than any other column next?

Latest [output](https://gist.github.com/hpoit/cb772dfd860731a46d95bb7955558272).

---

<div class="post-metadata">

**Author:** ![davidanthoff](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/davidanthoff/32/223493_2.png) [@davidanthoff](https://discourse.julialang.org/u/davidanthoff)\
**Post date:** [May 4, 2017, 5:27pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/33 "2017-05-04T17:27:24Z")

</div>

> [@hpoit](#):
>
> I had tried @select {i.Time\_G\_, i.Price, i.Volume} and that didn’t work so I didn’t think of @select {i.Price, i.Volume, i.Time\_G\_}. So the rule is place columns that apply to @where first in the list, than any other column next?

No, I think something else is mixed up, this should work with any ordering of the columns in the `@select` statement. Looking at your latest output, the CSV reading part seems to indicate that there are 0 rows, so you are running a query on a 0 rows data set. The problem seems to be somewhere upstream of the query, i.e. in the CSV reading part.

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 5:57pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/34 "2017-05-04T17:57:13Z")

</div>

This might be telling

```julia
julia> q = @from i in CSV.Source(io) begin
                  @where i.Type == "Trade"
                  @select {i.Price, i.Volume}
              end
ERROR: too many parameters for type _NT_
 in getiterator(::CSV.Source) at /Users/Corvus/.julia/v0.5/IterableTables/src/integrations/datastreams.jl:46
 in query(::CSV.Source) at /Users/Corvus/.julia/v0.5/Query/src/sources/source_iterable.jl:7

julia> 

```

---

<div class="post-metadata">

**Author:** ![davidanthoff](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/davidanthoff/32/223493_2.png) [@davidanthoff](https://discourse.julialang.org/u/davidanthoff)\
**Post date:** [May 4, 2017, 6:54pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/35 "2017-05-04T18:54:11Z")

</div>

I think again that something is going wrong with the CSV reading here, before the query (although Query should give a more helpful error message). Can you paste here somewhere what the `CSV.source(io)` expression returns?

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 8:01pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/36 "2017-05-04T20:01:24Z")

</div>

Absolutely

```julia
julia> CSV.Source(io)
CSV.Source: __IO__
    CSV.Options:
        delim: ','
        quotechar: '"'
        escapechar: '\\'
        null: ""
        dateformat: Base.Dates.DateFormat(Base.Dates.Slot[Base.Dates.DelimitedSlot{Base.Dates.Year}(Base.Dates.Year,'y',4,"-"),Base.Dates.DelimitedSlot{Base.Dates.Month}(Base.Dates.Month,'m',2,"-"),Base.Dates.DelimitedSlot{Base.Dates.Day}(Base.Dates.Day,'d',2,r"(?=\s|$)")],"","english")
Data.Schema{true}:
rows: 0	cols: 1
Columns:
 "" Nullable{WeakRefString{UInt8}}

julia> 

```

---

<div class="post-metadata">

**Author:** ![joshbode](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/joshbode/32/260_2.png) [@joshbode](https://discourse.julialang.org/u/joshbode)\
**Post date:** [May 4, 2017, 8:10pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/37 "2017-05-04T20:10:05Z")

</div>

Sorry, the `nb_available` for the Libz stream should have been `!eof`:

```julia
if !method_exists(nb_available, (BufferedStreams.BufferedInputStream{Libz.Source}, ))
    Base.nb_available{T <: Libz.Source}(s::BufferedStreams.BufferedInputStream{T}) = !eof(s)
end

```

It is only required to let `CSV.Source` know (once, at the start) that it can start on the stream and that it isn’t empty or already exhausted.

The `method_exists` check should make it a little bit more future-proof, in case an official method is later added.

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 8:45pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/38 "2017-05-04T20:45:47Z")

</div>

Thanks Josh. Here’s the [output](https://gist.github.com/hpoit/982776c868f421053889ef45e67fba65).

David, where should I apply `@time` in the [query definition](https://gist.github.com/hpoit/982776c868f421053889ef45e67fba65#file-pvt2-L12-L15)?

---

<div class="post-metadata">

**Author:** ![joshbode](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/joshbode/32/260_2.png) [@joshbode](https://discourse.julialang.org/u/joshbode)\
**Post date:** [May 4, 2017, 10:33pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/39 "2017-05-04T22:33:35Z")

</div>

You’ll need to either fix the header in the CSV file to exclude “non-identifier” characters (i.e. `Time[G] -> Time`), or use `getfield`, e.g.

```julia
q = @from i in CSV.Source(io) begin
    @where i.Type == "Trade"
    @select {i.Price, i.Volume, getfield(i, Symbol("Time[G]")}
end

```

---

<div class="post-metadata">

**Author:** ![hpoit](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/hpoit/32/245_2.png) [@hpoit](https://discourse.julialang.org/u/hpoit)\
**Post date:** [May 4, 2017, 11:18pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/40 "2017-05-04T23:18:15Z")

</div>

It’s actually `Time_G_`, which should be parsed as one identifier string just like `Price` and `Volume`?

I tried your snippet with `Time[G]` which didn’t work, and with `Time_G_` which [worked](https://gist.github.com/hpoit/cd63bca06d045660586f53a66390880c). In any case, I don’t understand the reason for this special treatment.

DataFrames is still not building.

---

<div class="post-metadata">

**Author:** ![joshbode](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/joshbode/32/260_2.png) [@joshbode](https://discourse.julialang.org/u/joshbode)\
**Post date:** [May 4, 2017, 11:22pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/41 "2017-05-04T23:22:07Z")

</div>

Hmm - Look Ma, no rows!

What do you get from:

```julia
io = Libz.ZlibInflateInputStream(open("file.csv.gz", "r"))
readline(io)

```

[Previous page](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502.md?page=1)

[Next page](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502.md?page=3)