使用 R 从 XML 中并行提取数据答案

【问题标题】：Extracting data from XML in parallel using R使用 R 从 XML 中并行提取数据
【发布时间】：2018-02-21 05:26:45
【问题描述】：

我每周必须处理数百万个 xml 文件，并且我使用具有 32 个内核的服务器。我在 Ubuntu 14.04 上运行 R。我当前的程序使用 for 循环从这些文件中提取数据并将输出保存到 MySQL。由于处理速度很慢，我尝试并行解析来自 xml 的数据。到目前为止，以下是我的代码：

//Function to parse data from xml

parse_xml <-function(FileName) {
  xmldata <- xmlParse(FileName)
  result_nodes = XML::getNodeSet(xmldata , "//IOTModellerLog/Event")
  if (!length(result_nodes)==0){
    abc <- rbindlist(lapply(result_nodes,function(x) data.frame(as.list(unlist(xmlToList(x))))),use.names = TRUE, fill = TRUE) 

....do other calculations....

    abc <- data.frame(lapply(abc, as.character), stringsAsFactors = FALSE)   
  }
}

//合并两个数据框的功能，NA填充不匹配的列

sbind = function(x, y, fill=NA) {
  sbind.fill = function(d, cols){ 
    for(c in cols)
      d[[c]] = fill
    d
  }
  x = sbind.fill(x, setdiff(names(y),names(x)))
  y = sbind.fill(y, setdiff(names(x),names(y)))

  rbind(x, y)
}

所有xml文件的路径

path <- "/home/mycomp/Documents/xmls/xmlfrommtar"

解析xml的函数。

workerFunc <- function(file) {
    dat <- parse_xml(file)
      df <- sbind(df,dat)
    }

并行处理代码

cl <- makeCluster(detectCores())
files <- dir(path, pattern="*.xml|*.XML", full.names = T)
part <- clusterSplit(cl, seq_along(files))
filesPart <- lapply(part, function(p) files[p])
results <- clusterApply(cl, filesPart, workerFunc)

当我运行它时，我收到以下错误：

Error in checkForRemoteErrors(val) : 
  4 nodes produced errors; first error: could not find function "parse_xml"

这对我来说表明了两件事，文件分布在我机器的所有 4 个内核上。函数 clusterApply 无法从 workerFunc 中找到 parse_xml。但是，我的环境中几乎有 parse_xml 文件。

示例 XML 数据

IOTModellerLog xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" DeviceID="7430180" ClientID="12324" FileCreationDate="2017-03-01T22:40:03" FileVersion="2" EventClassID="65535" IOTLogCreationDate="2017-03-01T12:29:54" SampleID="1" xsi:noNamespaceSchemaLocation="/opt/nds/ams_proxy/webapps/ams_proxy/WEB-INF/amsXmlSchema.xsd">
   <Event EventTime="2017-02-27T18:33:58">
      <IOTEvent State="PowerOn" />
   </Event>
   <Event EventTime="2017-02-28T08:59:03">
      <DataEvent>
         <Model>1</Model>
         <DataType>1</DataType>
         <DataValue>0301</DataValue>
      </DataEvent>
   </Event>
   <Event EventTime="2017-02-28T08:59:13">
      <DataEvent>
         <Model>1</Model>
         <DataType>1</DataType>
         <DataValue>0401</DataValue>
      </DataEvent>
   </Event>
</IOTModellerLog>

【问题讨论】：

我认为您需要使用 clusterExport 将函数导出到多个内核。

标签： r parallel-processing

【解决方案1】：

您应该使用以下内容初始化您的集群工作者：

clusterEvalQ(cl, library(XML))
clusterExport(cl, c('parse_xml', 'sbind'))

【讨论】：