Read and merge large tables on computer cluster

user_7742354 0
Viewed 141

I need to merge different large tables (up to 10Gb each) into a single one. To do so I am using a computer cluster with 50+ cores and 10+Gb Ram that runs on Linux.

I always end up with an error message like: "Cannot allocate vector of size X Mb". Given that commands like memory.limit(size=X) are Windows-specific and not accepted, I cannot find a way around to merge my large tables.

Any suggestion welcome!

This is the code I use:

library(parallel)

no_cores <- detectCores() - 1
cl <- makeCluster(no_cores)

temp = list.files(pattern="*.txt$")
gc()

Here the error occurs:

myfiles = parLapply(cl,temp, function(x) read.csv(x,
                                        header=TRUE, 
                                        sep=";",
                                        stringsAsFactors=F,
                                        encoding = "UTF-8",
                                        na.strings = c("NA","99","")))



myfiles.final = do.call(rbind, myfiles)
lapply parallel-processing cluster-computing merge
2 Answers

You could just use merge, for example:

`
mergedTable <- merge(table1, table2, by = "dbSNP_RSID")
If your samples have overlapping column names, then you'll find that the mergedTable has (for example) columns called Sample1.x and Sample1.y. This can be fixed by renaming the columns before or after the merge.

Reproducible example:

x <- data.frame(dbSNP_RSID = paste0("rs", sample(1e6, 1e5)),
  matrix(paste0(sample(c("A", "C", "T", "G"), 1e7, replace = TRUE),
    sample(c("A", "C", "T", "G"), 1e7, replace = TRUE)), ncol = 100))
y <- data.frame(dbSNP_RSID = paste0("rs", sample(1e6, 1e5)),
  matrix(paste0(sample(c("A", "C", "T", "G"), 1e7, replace = TRUE),
    sample(c("A", "C", "T", "G"), 1e7, replace = TRUE)), ncol = 100))
colnames(x)[2:101] <- paste0("Sample", 1:100)
colnames(y)[2:101] <- paste0("Sample", 101:200)
mergedDf <- merge(x, y, by = "dbSNP_RSID")

`

user_79648710

One way to approach this is with python and dask. The dask dataframe is stored mostly on disk rather than in ram- allowing you to work with larger than ram data- and can help you do computations with clever parallelization. A nice tutorial of ways to work on big data can be found in this kaggle post which might also be helpful for you. I also suggest checking out the docs on dask performance here. To be clear, if your data can fit in RAM using regular R dataframe or pandas dataframe will be faster.

Here's a dask solution which will assume you have named columns in the tables to align the concat operation. Please add to your question if you have any other special requirements about the data we need to consider. 

import dask.dataframe as dd
import glob

tmp = glob.glob("*.txt")

dfs= []
for f in tmp:
    # read the large tables
    ddf = dd.read_table(f)
    # make a list of all the dfs
    dfs.append(ddf)

#row-wise concat of the data
dd_all = dd.concat(dfs)
#repartition the df to 1 partition for saving
dd_all = dd_all.repartition(npartitions=1)

# save the data 
# provide list of one name if you don't want the partition number appended on
dd_all.to_csv(['all_big_files.tsv'], sep = '\t')

if you just wanted to cat all the tables together you can do something like this in straight python. (you could also use linux cat/paste). 

with open('all_big_files.tsv', 'w') as O:
    file_number = 0
    for f in tmp:
        with open(f, 'rU') as F:
            if file_number == 0:
                for line in F:
                    line = line.rstrip()
                    O.write(line + '\n')
            else:
                # skip the header line
                l = F.readline()
                for line in F:
                    line = line.rstrip()
                    O.write(line + '\n')
            file_number +=1
user_65397610
Related