大規模なcsvファイルをRで処理するストリーム

Question

非常に大きなcsvファイル（c.8.5GB）に対して比較的簡単な変更を加える必要があります。最初はさまざまなリーダー関数read.csv、readr :: read.csv、data.table :: freadを使って試してみました。しかし、それらはすべてメモリ不足です。

代わりにストリーム処理のアプローチを使用する必要があると思います。チャンクを読んだり、更新したり、書き込んだり、繰り返したりする。私は右の行にあるthis answerを見つけました。しかし、私はループを終了する方法はありません（私はRに比較的新しいです）。

だから私は2つの質問があります：

1. をwhileループの作業を行うための正しい方法は何ですか？
2. 「より良い」という定義にはより良い方法がありますか？例えばdplyr &パイプを使用してこれを行う方法がいくつかありますか？

現在のコードは次のように任意のポインタのための

src_fname <- "testdata/model_input.csv" 
tgt_fname <- "testdata/model_output.csv" 

#Changes needed in file: rebase identifiers, set another col to constant value 
rebase_data <- function(data, offset) { 
    data$'Unique Member ID' <- data$'Unique Member ID' - offset 
    data$'Client Name' <- "TestClient2" 
    return(data) 
} 

CHUNK_SIZE <- 1000 
src_conn = file(src_fname, "r") 
data <- read.csv(src_conn, nrows = CHUNK_SIZE, check.names=FALSE) 
cols <- colnames(data) 
offset <- data$'Unique Member ID'[1] - 1 

data <- rebase_data(data, offset) 
#1st time through, write the headers 
tgt_conn = file(tgt_fname, "w") 
write.csv(data,tgt_conn, row.names=FALSE) 

#loop over remaining data 
end = FALSE 
while(end == FALSE) { 
    data <- read.csv(src_conn, nrows = CHUNK_SIZE, check.names=FALSE, col.names = cols) 
    data <- rebase_data(data, offset) 
    #write.csv doesn't support col.names=FALSE; so use write.table which does 
    write.table(data, tgt_conn, row.names=FALSE, col.names=FALSE, sep=",") 
    # ??? How to test for EOF and set end = TRUE if so ??? 
    # This doesn't work, presumably because nrow() != CHUNK_SIZE on final loop? 
    if (nrow(data) < CHUNK_SIZE) { 
    end <- TRUE 
    } 

} 
close(src_conn) 
close(tgt_conn) 

感謝を。

Answer 1

OK次のように私は、解決策を見つけた：

# src_fname <- "testdata/model_input.csv" 
# tgt_fname <- "testdata/model_output.csv" 

CHUNK_SIZE <- 20000 

#Changes needed in file: rebase identifiers, set another col to constant value 
rebase_data <- function(data, offset) { 
    data$'Unique Member ID' <- data$'Unique Member ID' - offset 
    data$'Client Name' <- "TestClient2" 
    return(data) 
} 

#-------------------------------------------------------- 
# Get the structure first to speed things up 
#-------------------------------------------------------- 
structure <- read.csv(src_fname, nrows = 2, check.names = FALSE) 
cols <- colnames(structure) 
offset <- structure$'Unique Member ID'[1] - 1 

#Open the input & output files for reading & writing 
src_conn = file(src_fname, "r") 
tgt_conn = file(tgt_fname, "w") 

lines_read <- 0 
end <- FALSE 
read_header <- TRUE 
write_header <- TRUE 
while(end == FALSE) { 
    data <- read.csv(src_conn, nrows = CHUNK_SIZE, check.names=FALSE, col.names = cols, header = read_header) 
    if (nrow(data) > 0) { 
    lines_read <- lines_read + nrow(data) 
    print(paste0("lines read this chunk: ", nrow(data), ", lines read so far: ", lines_read)) 
    data <- rebase_data(data, offset) 
    #write.csv doesn't support col.names=FALSE; so use write.table which does 
    write.table(data, tgt_conn, row.names=FALSE, col.names=write_header, sep = ",") 
    } 
    if (nrow(data) < CHUNK_SIZE) { 
    end <- TRUE 
    } 
    read_header <- FALSE 
    write_header <- FALSE 
} 
close(src_conn) 
close(tgt_conn) 

Answer 2

はこれを試してみてください：

library("chunked") 

read_chunkwise(src_fname, chunk_size=CHUNK_SIZE) %>% 
rebase_data(offset) %>% 
write_chunkwise(tgt_fname) 

をあなたが正確に何をしたい得るためにCOLNAMESでビットをいじる必要があるかもしれません。

（免責事項：コードを試していない）：https://github.com/edwindj/chunked/