Rhadoop mapreduce для нескольких входных файлов

Question

Я строю программу mapreduce, используя R, которая извлекает соответствующие функции из набора функций в наборе данных, используя генетический алгоритм. Мне нужно поместить много файлов в качестве входных данных для моей работы mapreduce. Мой код ниже - это моя программа mapreduce, но она работает только для одного входного файла (data.csv).

library(caret)
library(dplyr)
library(rmr2)
Sys.setenv(HADOOP_CMD="/home/rania/hadoop-2.7.3/bin/hadoop")
Sys.getenv("HADOOP_CMD")
Sys.setenv(HADOOP_STREAMING="/home/rania/hadoop-streaming-2.7.3.jar")
library(rhdfs)
hdfs.init()
rmr.options(backend = "hadoop")
hdfs.mkdir("/user/rania/genetic")
hdfs.mkdir("/user/rania/genetic/data")

Я положил свои файлы в одну папку в формате hdfs.

hadoop fs -copyFromLocal /home/rania/Downloads/matrices/*.csv /user/rania/genetic/data/

Это функция карты

mon.map <- function(.,data){ 
data <- read.csv("/home/rania/Downloads/dataset.csv", header = T, sep = ";")
y <- c(1,0,1,0,1,1,1,1,0,0,1,0,1)

ga_ctrl <- gafsControl(functions = rfGA, # Assess fitness with RF
                       method = "cv")   # 10 fold cross validation
set.seed(10)
lev <- c("1","0")
rf_ga3 <- gafs(x = data, y = y,
                           iters = 10, # 100 generations of algorithm
                           popSize = 4, # population size for each generation
                           levels = lev,
                           gafsControl = ga_ctrl)
keyval(rf_ga3$ga$final, data[names(data) %in% rf_ga3$ga$final]  ) 
 }

Это функция уменьшения

mon.reduce <- function(k,v){
keyval(k,v) }

Теперь я применяю работу mapreduce

hdfs.root = 'genetic' 
hdfs.data = file.path(hdfs.root, 'data')
hdfs.out = file.path(hdfs.root, 'out')
csv.format <- make.output.format("csv")
genetic = function (input, output) {mapreduce(input=input, output=output, input.format="csv",output.format=csv.format, map=mon.map,reduce=mon.reduce)}
out = genetic(hdfs.data, hdfs.out)

Затем мы печатаем результат из hdfs

results <- from.dfs(out, format="csv")
print(results) 

ИЛИ ЖЕ

hdfs.cat("/genetic/out/part-00000")

Я попытался изменить функцию карты, чтобы она работала для многих файлов, но это не удалось

mon.map <- function(.,data){ 
data <- list.files(path="/home/rania/Downloads/matrices/", full.names=TRUE, pattern="\\.csv") %>% lapply(read.csv, header=TRUE, sep=",") 
y <- c(1,0,1,0,1,1,1,1,0,0,1,0,1)
for (i in 1:4){
ga_ctrl <- gafsControl(functions = rfGA, # Assess fitness with RF
                       method = "cv")   # 10 fold cross validation
set.seed(10)
lev <- c("1","0")
rf_ga3 <- gafs(x = data[[i]], y = y,
                           iters = 10, # 100 generations of algorithm
                           popSize = 4, # population size for each generation
                           levels = lev,
                           gafsControl = ga_ctrl)
               }
keyval(rf_ga3$ga$final, do.call(cbind, Map(`[`, data, c(rf_ga3$ga$final)))  )  
 }

Что я могу изменить в предыдущей функции карты, чтобы она работала для многих входных файлов? Спасибо