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

<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 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/1 "2017-05-03T16:30:48Z")

</div>

Hello! I’m not sure what the best way to sort is in this case but this is the plan: I would like to turn the below block of code into a loop (I was advised by @ChrisRackauckas not to loop through rows, and to use DataFramesMeta or Query)

```julia
using DataFrames
df = readtable(nrows=1000000, skipstart=0 "file1.csv")
sort!(df, cols = [:Type])
writetable("sorted_file1_1.csv", df)

```

The idea is to read every 1 million rows from the 5.2 GB CSV file, sort it by column `:Type`, do `writetable`, then repeat for the next million rows, by `readtable(nrows=1000000, skipstart=1000000, "file.csv")`.

Two variables I can think of right away are: `skipstart=` will increase by 1 million for each new file. `writetable` will increase the filename of sorted\_file1\_1 by 1 for every new file (e.g. the next file would be named `sorted_file1_2.csv`).

This approach is naive, because it doesn’t first consider the total amount of rows in the file, break it up into approximately equal size files, then sort all files.

After sorting, I need to read and input into a model only `df[:Type] .== "Trade"` rows, using `TextLineReader` with the `decode_csv` operation from Tensorflow.jl.

@SaschaMann @ScottPJones

---

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

</div>

If you only need a subset, why write out all the records? That seems inefficient. Also, if you sort by type, but then only keep one type, why sort?

In any case, when I deal with large files and only need a subset of the information, I use low-level routines from `CSV.jl`. You could read the file line by line, filter for `"Trade"`, keep what you need.

---

<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 3, 2017, 4: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/3 "2017-05-03T16:45:42Z")

</div>

Hi Tamas, thanks for the very good questions. Is there a way to read the 5 GB CSV file at once, or do I need to break it up into smaller files, and then read each file line by line, and filter for `"Trade"`? Let’s take it to the extreme, say I have a 30 GB CSV file?

---

<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 3, 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/4 "2017-05-03T17:03:25Z")

</div>

Just a warning, the sorting in [Query.jl](https://github.com/davidanthoff/Query.jl) is not very optimized at this point. I know how to improve it, just haven’t found the time yet.

In theory a [Query.jl](https://github.com/davidanthoff/Query.jl) query like this:

```julia
df = @from i in CSV.Source("file1.csv") begin
     @where i.Type=="Trade"
     @select i
     @collect DataFrame
end

```

should work with very large datasets because the whole design of both `DataStreams` and `Query` is one based on streaming data, i.e. this query will be executed without loading the whole content of `file1.csv` into main memory ever. As long as you filter out enough rows with the `@where` condition so that the resulting `DataFrame` fits into main memory, this should work in theory with an arbitrary large source csv file. Having said that, I have never tried it and this is how I _think_ things should work, reality unfortunately is often quite different. If you do try it, please report back if that works!

---

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

</div>

Thanks David for the script and tips. I’m about to try it now.

To limit `@where i.Type .== "Trade"` to 3000000 rows, should I do

`@where i.Type .== "Trade" && i < 3000000`?

Also, how do I initialize all of df without reading all of it?

---

<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 3, 2017, 6: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/7 "2017-05-03T18:18:27Z")

</div>

Tamas it’s bold to ask after receiving such wonderful insights from you, but could you provide example code for your suggested CSV.jl approach?

I’m skimming through [Home · CSV.jl](http://juliadata.github.io/CSV.jl/stable/#Lower-level-utilities-1) and assuming it would take a few numerous attempts to get this short block right.

---

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

</div>

I have handled 180GB CSV files on my laptop. The trick is to read line by line, discard what I don’t need, process in one or two passes. Also, compressing (`gz -9`) and reading from the stream saves a lot of space _and_ makes it faster (or at least not slower). The downside is that you need to write custom code.

---

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

</div>

Thanks Tamas. The per line read I need is very simple: if column `:Type` is `"Trade"`, store the `:Price` and `:Volume` element values, and discard everything else. I understand `CSV.readline` should be used, I’ll take a look at its source code.

I have the CSV file in GZ format as well, but how (with what tools/package) do I read from and write to a GZ? I found GZip.jl, would that work?

---

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

</div>

> [@hpoit](#):
>
> To limit `@where i.Type .== "Trade"` to 3000000 rows, should I do `@where i.Type .== "Trade" && i < 3000000`?

That won’t work, a `@where` clause is applied row by row, and the `i` in my example refers to one complete row, represented as a [NamedTuple](https://github.com/blackrock/NamedTuples.jl), so you can’t compare it to a number.

What you are looking for is the `@take` and `@skip` query operator, but both of them are not yet [implemented](https://github.com/davidanthoff/Query.jl/issues/68). They are on my long list of things to do.

In the meantime, the following might work:

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

df = DataFrame(take(q, 3_000_000))

```

The query in this example is _not_ terminated by a `@collect` statement. The whole query statement here constructs a query, but that query is not being run until a later time, namely when this query will be iterated. A query that ends with a `@collect` statement will run right away.

So `q` is essentially just a query that can be iterated. The `take` function is from julia base, and it creates a new iterator that will only iterate the first 3m elements from `q`. And then we pass that into the constructor of `DataFrames`, and thanks to [IterableTables.jl](https://github.com/davidanthoff/IterableTables.jl) this will materialize the query results into a `DataFrame`.

If you want to skip some rows in the beginning, you can use the `drop` function from julia base in a similar way.

At least in theory that query still should not load the whole file into main memory, so this might work with very large files, but again, I haven’t tested that.

Couple of other comments:

- I selected only two of the columns in this query, I took it from your later comments that you actually wanted to do that.
- I’m not sure what you want to do with the results of this query. Use the `DataFrame` right away with some other package? Or first write it out to disc? If the latter, you should actually not materialize things into a `DataFrame` at all, but instead write it directly to a CSV file, as described [here](https://github.com/davidanthoff/IterableTables.jl/blob/master/docs/src/userguide.md#datastreams-csv-feather). You would use code like `CSV.write("filename.csv", IterableTables.get_datastreams_source(take(q,3_000_000)))` for that.

> [@hpoit](#):
>
> Also, how do I initialize all of df without reading all of it?

I don’t understand that question, can you elaborate a bit?

---

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

</div>

Use the Libz package to read .gz files directly. In some tests I have done, it was only about 10% slower than reading the uncompressed files - an outstanding implementation!

---

<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 3, 2017, 9: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/12 "2017-05-03T21:56:16Z")

</div>

> [@davidanthoff](#):
>
> At least in theory that query still should not load the whole file into main memory, so this might work with very large files, but again, I haven’t tested that.

Hi David! I think this answers the question of initializing `df` without loading all of it into main memory. Thank you.

Yes, price and volume are all I need. Thank you!

The results of this query are supposed to be consumed by a TensorFlow.jl model in CSV format if I’m not mistaken, from this [documentation](https://www.tensorflow.org/programmers_guide/reading_data#reading_from_files). I haven’t even touched this topic of bridging between e.g. Query.jl and Tensorflow.jl, and am not sure if the latter will demand Query.jl to behave a certain way or not.

Also as Tamas and now recently @js135005 pointed out (thanks), can Query.jl work together with a GZ package to read directly from GZ files (which in this case would have to write to CSV for TensorFlow?)?

---

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

</div>

John, @Tamas_Papp said working with GZ files “makes it faster”. You’re saying it makes it slower. Which one is 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 3, 2017, 10:09pm UTC](https://discourse.julialang.org/t/how-to-subset-a-stream-i-o-and-pass-it-to-tensorflow-jl-question-has-been-updated/3502/14 "2017-05-03T22:09:26Z")

</div>

> [@hpoit](#):
>
> The results of this query are supposed to be consumed by a TensorFlow.jl model in CSV format if I’m not mistaken, from this documentation. I haven’t even touched this topic of bridging between e.g. Query.jl and Tensorflow.jl, and am not sure if the latter will demand Query.jl to behave a certain way or not.

I’m not familiar with TensorFlow, so I won’t be of much help on that count, but please report back! I guess I should at some point look into integrating [IterableTables.jl](https://github.com/davidanthoff/IterableTables.jl) with TensorFlow.jl, at that point queries would automatically work as input to TensorFlow.jl.

> [@hpoit](#):
>
> Also as Tamas and now recently js135005 pointed out (thanks), can Query.jl work together with a GZ package to read directly from GZ files (which in this case would have to write to CSV for TensorFlow?)?

That really depends on whether the [CSV.jl](https://github.com/JuliaData/CSV.jl) package works with one of these. In general you should be able to pass an `IO` stream instead of a filename to `CSV.Source`. If [GitHub - BioJulia/Libz.jl: Fast, flexible zlib bindings.](https://github.com/BioJulia/Libz.jl) returns an `IO` stream that works in that way with `CSV` it might all just work.

---

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

</div>

Okay David, here we go!

```julia
julia> q = @from i in CSV.Source("/file.csv") begin
           @where i.Type .== "Trade"
           @select {i.Price, i.Volumne}
       end
ERROR: UndefVarError: CSV not defined

julia> 

```

---

<div class="post-metadata">

**Author:** ![ChrisRackauckas](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/chrisrackauckas/32/77_2.png) [@ChrisRackauckas](https://discourse.julialang.org/u/ChrisRackauckas)\
**Post date:** [May 3, 2017, 11: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/16 "2017-05-03T23:28:09Z")

</div>

> [@hpoit](#):
>
> UndefVarError: CSV not defined

Did you not `using CSV`?

---

<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 3, 2017, 11: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/17 "2017-05-03T23:54:06Z")

</div>

Thanks Chris, for the millionth time.

Results for `.==`, [https://gist.github.com/hpoit/d6a5bf14e63091e94d7dc4f80f5496c6](https://gist.github.com/hpoit/d6a5bf14e63091e94d7dc4f80f5496c6)

Results for `==`, [https://gist.github.com/hpoit/af9774c7eb38c07d2fbf5fbd7ec2fa36](https://gist.github.com/hpoit/af9774c7eb38c07d2fbf5fbd7ec2fa36)

What’s more correct, `.==` or `==`, since `i` is already iterating over the row?

@davidanthoff

---

<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, 12: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/18 "2017-05-04T00:40:42Z")

</div>

You’ll also have to add `using DataFrames`.

You should use `==` for the comparison, all the expressions inside a query operate on a single row at a time.

`.==` will actually work in the next version of Query, at that point it will be the lifted version of the `==` operator that will return a `DataValue{Bool}`, whereas `==` will return just a `Bool`. So `.==` will propagate missing values, whereas `==` is going to be just a regular predicate that always returns either `true` or `false`.

---

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

</div>

Thanks David. [Here](https://gist.github.com/hpoit/fea6568a6fb2e21d7f364828d928c6f6) are the new results.

---

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

</div>

That’s exactly what Libz does. It provides a very efficient DataStream wrapper for gzip files. It looks just like a normal IO object so I see no reason why it shouldn’t work.

As I noted earlier, the overhead is very low and reading .gz files directly would certainly be less expensive than uncompressing them to clear files and then processing those clear files.

John Sutherland

---

<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: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/21 "2017-05-04T01:01:07Z")

</div>

Hi John, since I am for the moment working with historic data (and not real-time data), I’m not sure if an IO stream is necessary? Or should I just build an IO stream for historic and real-time data? Also, the historic data will come from one terminal, and the real-time, from another, so there will be two different layouts as well.

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