Processing Large CSV files with Elixir Streams

January 10, 2019

X Follow Button

We need to process a large CSV file of minute by minute volume and prices. Our task is pretty simple: we just want to get the first line of the year 2015, with valid data. At first, this seems an easy task we could tackle with String.split,Enum.map/filter/find. But what happens when the CSV file is large?

Let’s see how with Elixir Streams we can elegantly manage large files and create composable processing pipelines.

Getting a large CSV from Kaggle

We need at first a real and large CSV file to process and Kaggle is a great place where we can find this kind of data to play with. To download the CSV file just go to the Kaggle Bitcoin Historical Data page, and download the bitstampUSD CSV.

Kaggle – Historical BTC Page

The CSV is the historical Bitstamp BTC-USD prices and volumes aggregated with 1-minute interval. Once unzipped, the size is around 220Mbyte and it has 8 columns and 3.6 million rows.

Historical BTC-USD CSV

We will focus on the first two columns, Timestampand Open, filtering the header and all the rows where the open price is NaN. Converting Timestamp to a DateTime struct, we then find the first row of 2015 and return it.

The greedy approach

Elixir

Copy

The first column is the timestamp, where 1420070400 means 1st Jan 2015 00:00.

It works.. but this approach hides big memory and processing issues. With the Erlang Observer we can easily see the total memory allocated by our process.

iex> :observer.start

Elixir

Copy

Greedy approach – 5.5GB peak

Just processing a 220Mbyte CSV file, which is not even that big, we had a crazy 5.5GB peak of allocated memory.

Greedy Steps – Memory Allocation

Let’s inspect the code to understand what it does and how the data moves between each step.

line 1: File.read!("large.csv") loads the whole CSV file into memory. Just in this first step, we allocate 220Mbyte. And that’s just the beginning.

line 2: String.split(text,"\n") takes the loaded text and splits it into lines, creating a list of new strings representing the rows of the CSV.

line 3: Enum.map(rows, &String.split(&1, ",")) splits each single CSV row into a list of columns. The first column is the Timestamp, the second the Open price etc..

[\
  Timestamp,Open,High,Low,Close,...\
  ...\
  ["1420070400", "321", ...]\
  ...\
  ["1541888400","6359.96",..],\
  ["1541888460","NaN",..],\
  ...\
]

Elixir

Copy

line 4: We use Enum.filter/2 and pattern matching to filter out the header and rows with NaN values in the Open column.

[\
  ...\
  ["1420070400", "321", ...]\
  ...\
  ["1541888400","6359.96",..],\
  ["1541888520","6363.73",..],\
  ...\
]

Elixir

Copy

line 9: Enum.find/2 loops over the whole list of mapped and filtered rows, returning the result once the condition is matched.

["1420070400", "321", ...]

Elixir

Copy

We see how each one of these functions goes through the whole data set creating a new big data set. String.split splits the whole text into lines, Enum.map maps all the lines into columns etc… this is a huge waste of memory and processing resources.

Lazy Processing with Elixir Streams

We don’t need to load all the data in memory! We can actually try to load and process one line of text at a time. This is where Elixir Streams come into play!

Streams are composable, lazy enumerables

https://hexdocs.pm/elixir/Stream.html

Lazy means that when we use a Stream to process a CSV file, it doesn’t open and load the whole file. Instead, it reads one line at a time. We can compose complex pipelines with different processing steps where the stream reads and pipes one single line at a time, without having to process all the lines at every single step.

Elixir Streams – Looping – One line at a time

In the image above, we see how one single line is read and processed by the whole pipeline. Once the final step finishes to process it, a new line is then read and piped by the stream.

Instead of opening a file withFile.open, we use the functionFile.stream! to create a Stream.

iex> File.stream!("large.csv")
%File.Stream{
  line_or_bytes: :line,
  modes: [:raw, :read_ahead, :binary],
  path: "large.csv",
  raw: true
}

Elixir

Copy

File.stream!("large.csv")returns a Stream without opening the file. We can use this stream to compose a pipeline using the Stream module functions.

iex> File.stream!("large.csv") |> Stream.map(&String.split(&1,","))
#Stream<[\
  enum: %File.Stream{\
    line_or_bytes: :line,\
    modes: [:raw, :read_ahead, :binary],\
    path: "large.csv",\
    raw: true\
  },\
  funs: [#Function<48.51129937/1 in Stream.map/2>]\
]>

Elixir

Copy

Instead of usingEnum.map, we use Stream.map which returns a stream. This stream has a funs property which is a list of functions that will be applied to each row. At the moment there is no processing, we are only composing our pipeline.

To run our stream, we need to use a function that actually enumerate the stream, like Enum.count /take /find /map /filter etc..

iex> File.stream!("large.csv") |> Enum.count()
3603137

Elixir

Copy

As soon as Enum.count tries to loop through the stream, the stream opens the file and starts reading and passing the lines to Enum.count. If you look at the Erlang Observer, you’ll see that there is almost no memory peak, since the Enum.count function counts the lines one at the time

Compose our pipeline with streams

It’s now time to build our processing pipeline using Streams, instead of just Enum functions. You’ll see that the code will seem similar, but the way data flows between functions is quite different.

Let’s write the first part of the pipeline.

Elixir

Copy

#Stream<[\
  ...\
  funs: [#Function<48.51129937/1 in Stream.map/2>,\
   #Function<48.51129937/1 in Stream.map/2>,\
   #Function<40.51129937/1 in Stream.filter/2>]\
]>

Elixir

Copy

This first part of the pipeline returns a stream. The functions we pass to map and filter are the same as the initial example, and they are saved inside the stream processing pipeline. The first two map are where we trim each line and split it into columns. The third is where we filter out the header and NaN open prices.

Elixir

Copy

This is the final step, the one that actually runs the pipeline. The function we pass to Enum.find transforms the timestamp to a DateTime struct and returns true when it finds that the year is 2015.

Enum.find receives from the stream only the rows that are already filtered by the step before and once the condition dt.year == 2015 is met, it stops and returns the item.

filter -> 1370321820
find -> 1370321820 - 2013
filter -> 1370321880
find -> 1370321880 - 2013
...
filter -> 1420070340
find -> 1420070340 - 2014
filter -> 1420070400
find -> 1420070400 - 2015
["1420070400", "321", "321", "321",...]

Bash

Copy

Looking at the logs we’ve put in the filterandfind functions, we can see that each line is processed by the whole pipeline before starting to process the next one. Once the Enum.find function finds the first element with year 2015, it stops returning it, without having to process the rest of the stream.

Benchmarking with Benchee

Let’s compare the greedy and lazy approaches. We benchmark their speed with benchee while we monitor the memory footprint with the Erlang Observer.

Greedy VS Lazy

We see that the lazy approach has not only tremendous implications into memory consumption, but it also makes the code much faster compared to the greedy version.

tempest.services.disqus.com is blocked

This page has been blocked by an extension

Try disabling your extensions.

ERR_BLOCKED_BY_CLIENT

Reload

This page has been blocked by an extension

Disqus Comments

We were unable to load Disqus. If you are a moderator please see our troubleshooting guide.

Join the discussion…

Comment

Log in with

or sign up with Disqus or pick a name

Disqus is a discussion network

Don't be a jerk or do anything illegal. Everything is easier that way.

Read full terms and conditions

This comment platform is hosted by Disqus, Inc. I authorize Disqus and its affiliates to:

Use, sell, and share my information to enable me to use its comment services and for marketing purposes, including cross-context behavioral advertising, as described in our Terms of Service and Privacy Policy, including supplementing that information with other data about me, such as my browsing and location data.
Contact me or enable others to contact me by email with offers for goods or services
Process any sensitive personal information that I submit in a comment. See our Privacy Policy for more information

Acknowledge I am 18 or older

7
Discussion Favorited!

Favoriting means this is a discussion worth sharing. It gets shared to your followers' Disqus feeds, and gives the creator kudos!

Find More Discussions

Tweet this discussion
- Share this discussion on Facebook
- Share this discussion via email
- Copy link to discussion
Best
- Newest
- Oldest

What about newlines inside of a single row or quoted commas? this isn't exactly valid CSV parsing

the following is a perfectly valid **2** line csv

" asdf, still column 1,

still column1 line 1, asdf",asdf,asdf

" asdf

asdf",asdf,asdf

Hey! Yes, this would not work with newlines. This example is focused on Streams and for simplicity shows a csv file with a specific type of data. In general I would use something like https://hexdocs.pm/nimble_c... which is really fast!

Good enough for me

Show more replies

Great post! I need to check Elixir's Streams for XML processing.

Now you might want to discover Flow and make it 8 times faster (assuming your machine has 8 cores.)

Also, stream CSV processor NimbleCSV might make the processing even faster by times.

Thanks for your comment and linking the resources! Yep Flow is fantastic, especially if the data can be divided and processed in different processes! It's awesome how it integrates producer/consumer paradigm inside a stream pipeline.

The point of this particular article is to show lazy processing with streams, though.

Great Post.

I am doing some lab using Elixir to process 140 milion of lines. But I dont have any success to perform a fast read of file.

Stream seems very slow when comparing Elixir time ( about 5 min ) to Go (23 seconds) or Rust (13 seconds)

Its not clear yet why or where is the slower part of process, but im having a lot of fun with the elixir approach

Hey Alam,

Go and Rust are supposed to have better performance on this kind of task. But I think you can maybe make some improvements. Can you please share your code and the 140M lines (or a similar file which I can run a benchmark on)?

Hi Alvise,

Thank you for share your time.

Before we can start, this was just a simple Elixir test. I made this kind of program in Java in my Job and i was curious about how Elixir could perform some IO Operation.

You can download the files here : http://www.portaltransparencia.gov.br/download-de-dados/bolsa-familia-pagamentos

I made some file merges to get 140 M, but if you download a month like September ( Setembro ) you will catch ~13M in a file.

My first attempt was :

def load_stream file do begin_time = NaiveDateTime.utc_now data = file |> File.stream!() |> CSV.decode(separator: ?;, headers: [:Mesreferencia, :Mescompetencia, :Uf, :Codigomunicipiosiafi, :Nomemunicipio, :Cpffavorecido, :Nisfavorecido, :Nomefavorecido, :Valorparcela]) |> Enum.to_list end_time = NaiveDateTime.utc_now {:ok, begin_time, end_time, data} end

Note : the CSV.decode line uses a diferent headers name. You have to change the headers file if you want to use this.

Note2: CSV uses this dependency : {:csv, "~> 2.4.1"}

Note3: Enum.to_list was necessary cause i was trying to see how much memory would be needed. During my tests, September File uses 6.5 Gb In Go and 7gb in Rust. But in rust the number becomes 3.5gb right after the end of stream,

Well, I see the NimbleCSV tip in your thread and I decide remove the CSV.decode line to avoid unecessary overhead and for my surprise, the process take a lot of time to end.

Just for curiosity, Looking :observer.start I can see a lot of parallel process running with the main process and i wondered if there is a way to avoid it. I just dont know how can i make it.

I pushed this tiny code here, if you want. But for now, its just the same code above

https://github.com/xawe/Elixir_CSV_Loader.git

I gave it a try with a 14M lines CSV. Don't use `Enum.to_list/1` at the end, because you loose the advantage of using streams. The file is too big to be processed as a single list.

By using `:csv` library, with your code, it takes 6 minutes to convert map all the 14M lines to Elixir maps. With nimble_csv it's much faster, around 1m10s. The code is similar:

` NimbleCSV.define(MyParser, separator: ";", escape: """)

def load_stream file do file |> File.stream!() |> MyParser.parse_stream() |> Enum.count() end `

Then you run the code:

$ iex -S mix iex> :timer.tc fn -> CsvLoader.load_stream("202009_BolsaFamilia_Pagamentos.csv") end {93818429, 14278284}

93818429 are the microseconds (1m30s), 14278284 the number of streamed maps.

With Flow it's even faster (and you'll appreciate even more this library if you do real processing apart from converting the csv lines into maps), it takes 57s

` NimbleCSV.define(MyParser, separator: ";", escape: """)

def load_stream file do file |> File.stream!() |> Flow.from_enumerable() |> Flow.map(&MyParser.parse_string/1) |> Enum.count() end `

Thanks Alam, going to try this tomorrow :D Can you please write me an email to alvise@poeticoding.com, so we can continue there?

Load more comments

live.rezync.com