Читать и объединять большие таблицы на компьютерном кластере

Question

Мне нужно объединить разные большие таблицы (до 10 Гб каждая) в одну. Для этого я использую компьютерный кластер с 50+ ядрами и 10 + Gb Ram, который работает в Linux.

Я всегда получаю сообщение об ошибке вроде: «Невозможно выделить вектор размером X Мб». Учитывая, что такие команды, как memory.limit(size=X), специфичны для Windows и не принимаются, я не могу найти способ объединить свои большие таблицы.

Любое предложение приветствуется!

Это код, который я использую:

library(parallel)

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

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

Здесь возникает ошибка:

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)

Answer 1

Один из способов подойти к этому с помощью python и dask. Фрейм данных dask хранится в основном на диске, а не в оперативной памяти, что позволяет вам работать с данными больше оперативной памяти, и может помочь вам выполнить вычисления с умным распараллеливанием. Хорошее руководство по работе с большими данными можно найти в этом посте kaggle , который также может быть полезен для вас. Я также предлагаю ознакомиться с документацией по производительности dask здесь . Чтобы было ясно, если ваши данные могут поместиться в ОЗУ, используя обычный R-фрейм данных или Pandas, фрейм данных будет быстрее.

Вот решение dask, которое предполагает, что вы назвали столбцы в таблицах для выравнивания операции concat. Пожалуйста, добавьте к своему вопросу, если у вас есть какие-то особые требования к данным, которые мы должны рассмотреть.

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')

Если вы просто хотите объединить все столы вместе, вы можете сделать что-то подобное в прямом питоне. (вы также можете использовать 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

Answer 2

Вы можете просто использовать слияние, например:

`
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")

`