# 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:** 3

<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: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/42 "2017-05-04T23:27:25Z")

</div>

Lol. I get

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

julia> readline(io)
"#RIC,Date[G],Time[G],GMT Offset,Type,Price,Volume,Bid Price,Bid Size,Ask Price,Ask Size,Qualifiers\n"

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, 11:39pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/43 "2017-05-04T23:39:40Z")

</div>

OK - looks right.

All I can think of is that you need to make sure `io` hasn’t already been used before running the query, otherwise it will have been exhausted (or missing some rows, at least, including the header) - i.e. don’t rerun the query `q` without reopening `io`.

---

<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 5, 2017, 12: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/44 "2017-05-05T00:10:20Z")

</div>

[pvt3](https://gist.github.com/hpoit/cd63bca06d045660586f53a66390880c) was run once from a new Julia terminal. David, I still need your help.

---

<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 5, 2017, 4:45am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/45 "2017-05-05T04:45:09Z")

</div>

I’m not sure, as I said, this looks like a problem with reading the CSV file, not with the query part of things. If you look at the pvt3 output you can see a line that says `rows: 0	cols: 1`. That info comes from the CSV.jl package and means that the csv reading code for some reason sees 0 rows and only one column in the file. pvt2 on the other hand doesn’t have that problem, so you are changing something in the way you read the CSV file there, all before the query part.

---

<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 5, 2017, 7:01am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/46 "2017-05-05T07:01:58Z")

</div>

> [@hpoit](#):
>
> where should I place @time?

Normally one would not benchmark in global scope, but in your case the runtime should dominate compilation on the full dataset, so it should be OK.

---

<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 5, 2017, 11:04am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/49 "2017-05-05T11:04:20Z")

</div>

> [@davidanthoff](#):
>
> That info comes from the CSV.jl package and means that the csv reading code for some reason sees 0 rows and only one column in the file.

Okay, CSV.jl on pvt2 “sees” right, but then what’s happening on the next step, [`df = DataFrame(take(q, 10_000_000))`](https://gist.github.com/hpoit/982776c868f421053889ef45e67fba65#file-pvt2-L53)?

---

<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 5, 2017, 11:23am UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/50 "2017-05-05T11:23:47Z")

</div>

> [@Tamas\_Papp](#):
>
> Normally one would not benchmark in global scope, but in your case the runtime should dominate compilation on the full dataset, so it should be OK.

Well, the benchmark runs okay  
[`julia> @time query() 196.525924 seconds (5.28 M allocations: 8.289 GB, 1.23% gc time)`](https://gist.github.com/hpoit/bdb11a4aaf0b689c4aa8efff284c9c62#file-pvt2-with-time-L30-L31)

but now [`df = DataFrame(take(query(), 10_000_000))`](https://gist.github.com/hpoit/bdb11a4aaf0b689c4aa8efff284c9c62#file-pvt2-with-time-L69) won’t start because it can’t take in `query()`.

---

<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 5, 2017, 3:00pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/51 "2017-05-05T15:00:31Z")

</div>

> [@davidanthoff](#):
>
> I’m not sure, as I said, this looks like a problem with reading the CSV file, not with the query part of things. If you look at the pvt3 output you can see a line that says rows: 0 cols: 1. That info comes from the CSV.jl package and means that the csv reading code for some reason sees 0 rows and only one column in the file. pvt2 on the other hand doesn’t have that problem, so you are changing something in the way you read the CSV file there, all before the query part.

Hi @quinnj, is the problem inside CSV.jl? If yes, where do I start to learn about it (/Source.jl or /io.jl)? Thanks.

---

<div class="post-metadata">

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

</div>

> [@hpoit](#):
>
> “#RIC,Date[G],Time[G],GMT Offset,Type,Price,Volume,Bid Price,Bid Size,Ask Price,Ask Size,Qualifiers\n”

Umm…not exactly sure. What’s the csv file in question? Can it be posted/attached here so I can take a look? Along with the `CSV.Source` or `CSV.read` command being used?

---

<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 5, 2017, 6:23pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/53 "2017-05-05T18:23:49Z")

</div>

> [@hpoit](#):
>
> Well, the benchmark runs okay  
> julia\> @time query()  
> 196.525924 seconds (5.28 M allocations: 8.289 GB, 1.23% gc time)
> 
> but now df = DataFrame(take(query(), 10\_000\_000)) won’t start because it can’t take in query().

You are running into the same problem that @joshbode described already:

> [@joshbode](#):
>
> All I can think of is that you need to make sure io hasn’t already been used before running the query, otherwise it will have been exhausted (or missing some rows, at least, including the header) - i.e. don’t rerun the query q without reopening io.

You can’t call `CSV.Source` twice on the same io stream. I’m not sure whether the stream from libz can somehow be reset after you’ve read from it once, or whether you’ll just have to open a new stream for the second call to `CSV.Source`.

---

<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 5, 2017, 6:59pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/54 "2017-05-05T18:59:33Z")

</div>

Can resetting still be the problem?

```julia
julia> CSV.reset!(CSV.Source(io))

julia> df = DataFrame(take(query(), 10_000_000))
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
 in query() at ./REPL[4]:2

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 5, 2017, 7:19pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/55 "2017-05-05T19:19:38Z")

</div>

You would need to reset the io stream, not the `CSV.Source` instance. I think `seekstart(io)` before you call `CSV.Source(io)` might do the trick, if libz streams support that.

---

<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 5, 2017, 7:24pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/56 "2017-05-05T19:24:45Z")

</div>

Hi Jacob, I wasn’t able to attach a CSV here, but here’s a picture of a sample of it

 ![](https://global.discourse-cdn.com/julialang/original/3X/7/a/7a60989a509eafc690f512e5327d2e0cac17e8fa.png)

Source and read commands used are

```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> CSV.read(io)
0×1 DataFrames.DataFrame

julia> 

```

Perhaps we’re missing or mis-specified some `CSV.Options` or `Data.Schema`? Or maybe the I/O changes the reading? It was reading fine as a [CSV](https://gist.github.com/hpoit/6b26f41a12fc544fa97bb6d8c19f8c5a#file-pv2-L63) (not [I/O)](https://gist.github.com/hpoit/bdb11a4aaf0b689c4aa8efff284c9c62#file-pvt2-with-time-L23).

---

<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 5, 2017, 7:46pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/57 "2017-05-05T19:46:07Z")

</div>

I couldn’t find `seekstart()` in Libz.jl.

---

<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 5, 2017, 7:46pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/58 "2017-05-05T19:46:55Z")

</div>

It is very clear what is going on here: you are reading repeatedly from the `io` stream without calling `seekstart` on the stream. The example you just posted is another case of that: You can’t call `CSV.Source(io)` followed by `CSV.read(io)` without calling `seekstart(io)` inbetween.

If `seekstart` is not supported by libz you will have to reopen the stream once you have read from it via `CSV.Source` or `CSV.read` or anything like that.

---

<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 5, 2017, 7: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/59 "2017-05-05T19:54:52Z")

</div>

I couldn’t find it in Libz.jl but does this mean Libz.jl still supports it?

```julia
help?> seekstart
search: seekstart

  seekstart(s)

  Seek a stream to its beginning.

julia> 

```

In any case, this is the sequence

```julia
using CSV, DataFrames, Query, Libz

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

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

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

@time query()

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

```

How would I define a `seekstart(io)` call before the second call to `CSV.Source(io)`?

---

<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 5, 2017, 7: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/60 "2017-05-05T19:56:58Z")

</div>

Oh I got this as well, does that mean Libz does not support it?

```julia
julia> seekstart(io)
ERROR: ArgumentError: Can't seek in input stream with source of type Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}
 in seek(::BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}}, ::Int64) at /Users/Corvus/.julia/v0.5/BufferedStreams/src/bufferedinputstream.jl:323
 in seekstart(::BufferedStreams.BufferedInputStream{Libz.Source{:inflate,BufferedStreams.BufferedInputStream{IOStream}}}) at ./iostream.jl:54

julia> 

```

---

<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 5, 2017, 8:05pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/61 "2017-05-05T20:05:04Z")

</div>

> [@davidanthoff](#):
>
> If seekstart is not supported by libz you will have to reopen the stream once you have read from it via CSV.Source or CSV.read or anything like that.

Okay let’s assume it does not support it, how do I reopen I/O with `CSV.Source()` or `CSV.read()` before the second call to `CSV.Source(io)`?

---

<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 5, 2017, 8:16pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/62 "2017-05-05T20:16:45Z")

</div>

[Here](https://gist.github.com/hpoit/249d3d630c17563c2bc4fcd4be01d6f4)’s the latest application of the sequence with `seekstart(io)`, before I called `CSV.Source(io)` for the second time.

Hello @bicycle1885 and @dcjones, would you know where I could get instructions to support Libz.jl with `seekstart(io)`?

---

<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 5, 2017, 8:23pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/63 "2017-05-05T20:23:55Z")

</div>

Just put it all into a function like this:

```julia
function query(filename)
    file = open(filename, "r")
    io = Libz.ZlibInflateInputStream(file)
    csv = CSV.Source(io)
    q = @from i in csv begin
        @where i.Type == "Trade"
        @select {i.Price, i.Volume, i.Time_G_}
    end
    df = DataFrame(take(query(), 10_000_000))
    return df
end

```

Note a couple of things:

- If you use the `@time` macro make sure you run the function once before you time it, otherwise you are just timing the compiler.
- This function is now opening lots of things that should be closed after use (file stream, probably libz stream, and I think the CSV.Source instance). Ideally you would use `try ... finally` blocks to make sure these resources always get freed. Something like this:

```julia
function query(filename)
    file = open(filename, "r")
    try
        io = Libz.ZlibInflateInputStream(file)
        try
            csv = CSV.Source(io)
            try
                q = @from i in csv begin
                    @where i.Type == "Trade"
                    @select {i.Price, i.Volume, i.Time_G_}
                end
                df = DataFrame(take(query(), 10_000_000))
                return df
            finally
                Data.close!(csv)
            end
        finally
            close(io)
        end
    finally
        close(file)
    end
end

```

[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=2)

[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=4)