Import csv and merge

library(disk.frame)

I chose 128 chunks because, 10 years of data, 12 months each (except 1987), roughly 120. Let’s round it to nearest 2^n.

Actually, what I intent to do was, chunk each data into arbitrary number but at the end, during rbindlist.disk.frame I should be saying bind to 128 chunks, but rbindlist.disk.frame didn’t have such option.

On the other hand, If I chose 12 chunks at csv_to_disk.frame stage then I might have force rbindlist.disk.frame merging data into 12 chunks. Thus I picked 128 at the beginning.

list.files("original_data", 
           pattern="*.csv", 
           full.names = T) %>% 
  head(10) %>%  # only 10 files so that we can test faster
  purrr::map(~ csv_to_disk.frame(.x, 
                                outdir=paste0("tmp", readr::parse_number(.x)),
                                nchunks = 128,
                                overwrite=T)) %>% 
  rbindlist.disk.frame(outdir = "flights_all_notsharded.df", by_chunk_id = T)

Now each (temporary) folder has 128 files:

$ find -iname "*.fst" | grep tmp | cut -d'/' -f2 | sort | uniq -c
    128 tmp1987
    128 tmp1988
    128 tmp1989
    128 tmp1990
    128 tmp1991
    128 tmp1992
    128 tmp1993
    128 tmp1994
    128 tmp1995
    128 tmp1996

Clean up directories for individual file imports (I wish there was no need for duplication of folders when rbindlist.disk.frame is used)

rm -r tmp????

Use the flight data

flights_df <- disk.frame("flights_all_notsharded.df/")
flights_df

## path: "flights_all_notsharded.df/"
## nchunks: 128
## nrow: 47925064
## ncol: 29

No missing files in merged folder

ls flights_all_notsharded.df/ | wc -l

## 128

two-stage group_by

flights_df %>%
  group_by(UniqueCarrier) %>% # per chunk
  summarize(count = n()) %>%  
  collect %>%  
  group_by(UniqueCarrier) %>% # after collect
  summarize(count = sum(count)) %>% 
  arrange(UniqueCarrier) %>% 
  knitr::kable()  # somehow the output was problematic to print in html, thus used kable

by year

Let’s do counts per year this time.

flights_df %>%
  group_by(Year) %>% # per chunk
  summarize(count = n()) %>%  
  collect %>%  
  group_by(Year) %>% # after collect
  summarize(count = sum(count)) %>% 
  arrange(Year) %>% 
  knitr::kable()

Here’s combined table (note1: actual data has one extra line, the header line, per file. note2: the current results is indicated with not_sharded).

Actual counts have one additional line per file, thus the total has extra 10 lines extra, so there’s no discrepancy

per chunk operation

Let’s calculate total number of rows

delayed(flights_df, ~nrow(.x)) %>% collect_list() %>% purrr::reduce(`+`)

## [1] 47925064

Here’s the imap approach.. No problems..

disk.frame::imap.disk.frame(flights_df, ~nrow(.x), lazy = F) %>% purrr::reduce(`+`)

## [1] 47925064