java - 使用库(并行)在 R for Windows 上对 RWeka 进行并行化

Question

由于 mclapply() 不可用，我正在使用库(并行)来分发大型语料库的 NER。我正在使用this lmullen 提供的很好的教程作为起点。

首先，编写这样的代码片段:

annotate_entities <- function(doc, annotation_pipeline) {
    annotations <- annotate(doc, annotation_pipeline)
    AnnotatedPlainTextDocument(doc, annotations)
}

然后，管道:

itinerants_pipeline <- list(
  Maxent_Sent_Token_Annotator(),
  Maxent_Word_Token_Annotator(),
  Maxent_Entity_Annotator(kind = "person"),
  Maxent_Entity_Annotator(kind = "location")
)

语料库处理作品的串行版本，即

 corpus_serial_NER <- lapply(corpus[1:100]$content, annotate_entities, itinerants_pipeline ) 

但是，当我尝试并行化时，我遇到了麻烦:

library(parallel)

cl <- makePSOCKcluster(8)
setDefaultCluster(cl)

clusterExport(cl, c('annotate_entities', 
 'annotate', 
 'Maxent_Sent_Token_Annotator', 
 'AnnotatedPlainTextDocument', 
 'Maxent_Sent_Token_Annotator', 
  'Maxent_Word_Token_Annotator',
  'Maxent_Entity_Annotator'))

corpus_NER <- parLapply(cl, corpus[1:100]$content, function(k) {
  itinerants_pipeline <- list(
  Maxent_Sent_Token_Annotator(),
  Maxent_Word_Token_Annotator(),
  Maxent_Entity_Annotator(kind = "person"),
  Maxent_Entity_Annotator(kind = "location"))

  annotate_entities(k, itinerants_pipeline)
  }))

如果我尝试仅导出上面的函数而没有“内部”，引擎会报告它们丢失。四处搜索给我的印象是，这是因为当函数进入并行化时，对 Java 对象的引用被“切断”。但我怀疑对此的处理让我感到悲伤。

对于很小的语料库(10 个文档)，但超过 50 个，它会崩溃并显示以下消息:

Error in checkForRemoteErrors(val) : 
  7 nodes produced errors; first error: java.lang.OutOfMemoryError: GC overhead limit exceeded
In addition: 
Warning messages:
1: 
In .Internal(gc(verbose, reset)) :


 closing unused connection 14 (<-NO92W:11748)
2: 
In .Internal(gc(verbose, reset)) : (.................)

我了解到此错误消息来自 Java，并且与过度垃圾收集有关。但是，我不明白是什么导致在我的并行代码中发生这种情况(而不是在我串行运行它时)。

我想知道是什么原因导致了这种行为，但我也对解决方法感兴趣。我不清楚在 R/Windows 上进行并行 lapply 的最佳方法是什么，但使用了这个解决方案，因为我能够让它与其他函数(那些不是来自 Java 的)一起工作。

Answer 1

我发布一个答案是因为它为上述确切问题提供了一个可行的解决方案，感谢 Roman Luštrik's评论。加载包，然后删除函数中管道的创建解决了这个问题。这是工作代码:

cl <- makePSOCKcluster(7)
setDefaultCluster(cl)

clusterEvalQ(cl, library(NLP));
clusterEvalQ(cl, library(openNLP));
clusterEvalQ(cl, library(RWeka));
clusterEvalQ(cl, library(openNLPmodels.en));

clusterEvalQ(cl, itinerants_pipeline <- list(
    Maxent_Sent_Token_Annotator(),
    Maxent_Word_Token_Annotator(),
    Maxent_Entity_Annotator(kind = "person"),
    Maxent_Entity_Annotator(kind = "location")))

clusterExport(cl, c('annotate_entities'))

system.time(corpus_par_NER <- parLapply(cl,corpus[1:5000]$content, function(k) {
    annotate_entities(k, itinerants_pipeline)
}))

stopCluster(cl)

至关重要的是，管道以这种方式导出。通过 clusterExport 执行此操作(与“annotate_entities”在同一列表中不起作用)。