Использование списков в foreach R

Я пытаюсь распараллелить извлечение данных, сохраненных в некоторых html-документах, и сохранить их в data.frames(некоторые миллионы документов, следовательно, полезность распараллеливания).

На первом этапе, на машине, где я регистрирую очередь, я выбираю подмножество html файлов и привязываю к ним функцию read_html (из пакета rvest я также пробовал аналогичную функцию из пакета XML но я получал проблемы с утечкой памяти), чтобы получить уникальный список, хранящий содержимое многих html-страниц.

Затем я использую итератор в этом списке, чтобы получить меньшие куски его для подачи в foreach.

Внутри foreach я создаю data.frame(s) (используя функцию html_table и некоторые базовые манипуляции с данными), и я возвращаю список, элементами которого являются очищенные data.frames.

Я попытался использовать бэкэнд doSNOW на win 8 и doRedis на ubuntu 16.04.

В первом случае возвращается список пустых списков, а во втором - ошибка отображения памяти; вы можете найти трассировку в самом низу вопроса.

По моему мнению, списки (куски), которые я отправляю на ядра, не ведут себя так, как я ожидаю. Я собрал вокруг, что объект списка может быть всего лишь набором указателей, но я не смог его подтвердить; может быть, это может быть проблемой? Есть ли альтернатива "способу списка" для "инкапсулирования" данных с несколькими страницами html?

Ниже вы можете найти код, воспроизводящий проблему. Я совершенно новый для, новый для параллельного программирования и довольно новый для программирования R: любые советы по улучшению приветствуются. Спасибо всем заранее.

library(rvest)
library(foreach)

#wikipedia pages of olympic medalist between 1992 and 2016 are
# downloaded for reproducibility
for(i in seq(1992, 2016, by=4)){

  html = paste("https://en.wikipedia.org/wiki/List_of_", i, "_Summer_Olympics_medal_winners", sep="")
  con = url(html)
  htmlCode = readLines(con)
  writeLines(htmlCode, con=paste(i, "medalists", sep="_"))
  close(con)

}

#declaring the redis backend (doSNOW code is also included below)

#note that I am using the package from 
#devtools::install_github("bwlewis/doRedis") due to a "nodelay error"
#(more info on that here: https://github.com/bwlewis/doRedis/issues/24)
# if it is not your case please drop the nodelay and timeout options

#Registering cores ---Ubuntu---
cores=2
library('doRedis')
options('redis:num'=TRUE)
registerDoRedis("jobs", nodelay=FALSE)
startLocalWorkers(n=cores, "jobs", timeout=2, nodelay=FALSE)
foreachOpt <- list(preschedule=FALSE)


#Registering cores ---Win---
#cores=2
#library("doSNOW")
#registerDoSNOW(makeCluster(cores, type = "SOCK"))


#defining the iterator
iterator <- function(x, ...) {
  i <- 1
  it <- idiv(length(x), ...)

  if(exists("chunks")){
    nextEl <- function() {
      n <- nextElem(it)
      ix <- seq(i, length=n)
      i <<- i + n
      x[ix]
    }
  }else{
    nextEl <- function() {
      n <- nextElem(it)
      ix <- seq(i, i+n-1)
      i <<- i + n
      x[ix]
    }
  }
  obj <- list(nextElem=nextEl)
  class(obj) <- c(
    'ivector', 'abstractiter','iter')
  obj
}

#reading files
names_files<-list.files()
html_list<-lapply(names_files, read_html)

#creating iterator
ChunkSize_html_list<-2
iter<-iterator(html_list, chunkSize=ChunkSize_html_list)

#defining expanding list (thanks StackOverflow and many thanks to
#JanKanis answer : http://stackoverflow.com/info/2436688/append-an-object-to-a-list-in-r-in-amortized-constant-time-o1  )
expanding_list <- function(capacity = 10) {
  buffer <- vector('list', capacity)
  length <- 0

  methods <- list()

  methods$double.size <- function() {
    buffer <<- c(buffer, vector('list', capacity))
    capacity <<- capacity * 2
  }

  methods$add <- function(val) {
    if(length == capacity) {
      methods$double.size()
    }

    length <<- length + 1
    buffer[[length]] <<- val
  }

  methods$as.list <- function() {
    b <- buffer[0:length]
    return(b)
  }

  methods
}

#parallelized part
clean_data<-foreach(ite=iter, .packages=c("itertools", "rvest"), .combine=c,
 .options.multicore=foreachOpt, .options.redis=list(chunkSize=1)) %dopar% {

  temp_tot <- expanding_list()
      for(g in 1:length(ite)){

        #extraction of data from tables
      tables <- html_table(ite[[g]], fill=T, header = T)

        for(i in 1:length(tables)){

          #just some basic data manipulation
          temp<-lapply(tables, function(d){d[nrow(d),]})
          temp_tot$add(temp)
          rm(temp)
          gc(verbose = F)
        }
      }
  #returning the list of cleaned data.frames to the foreach 
    temp_tot$as.list()
}

Ошибка при использовании backis-сервера redis:

*** caught segfault ***
address 0x60, cause 'memory not mapped'


Traceback:
 1: .Call("xml2_doc_namespaces", PACKAGE = "xml2", doc)
 2: doc_namespaces(doc)
 3: xml_ns.xml_document(x)
 4: xml_ns(x)
 5: xpath_search(x$node, x$doc, xpath = xpath, nsMap = ns, num_results = Inf)
 6: xml_find_all.xml_node(x, ".//table")
 7: xml2::xml_find_all(x, ".//table")
 8: html_table.xml_document(ite[[g]], fill = T, header = T)
 9: html_table(ite[[g]], fill = T, header = T)
10: eval(expr, envir, enclos)
11: eval(.doRedisGlobals$expr, envir = .doRedisGlobals$exportenv)
12: doTryCatch(return(expr), name, parentenv, handler)
13: tryCatchOne(expr, names, parentenv, handlers[[1L]])
14: tryCatchList(expr, classes, parentenv, handlers)
15: tryCatch({    lapply(names(args), function(n) assign(n, args[[n]], pos = .doRedisGlobals$exportenv))    if (exists(".Random.seed", envir = .doRedisGlobals$exportenv)) {        assign(".Random.seed", .doRedisGlobals$exportenv$.Random.seed,             envir = globalenv())    }    tryCatch({        if (exists("set.seed.worker", envir = .doRedisGlobals$exportenv))             do.call("set.seed.worker", list(0), envir = .doRedisGlobals$exportenv)    }, error = function(e) cat(as.character(e), "\n"))    eval(.doRedisGlobals$expr, envir = .doRedisGlobals$exportenv)}, error = function(e) e)
16: FUN(X[[i]], ...)
17: lapply(work[[1]]$argsList, .evalWrapper)
18: redisWorker(queue = "jobs", host = "localhost", port = 6379,     iter = Inf, linger = 30, log = stdout(), timeout = 2, nodelay = FALSE)
aborting ...

Ответ 1