parallel ensembles

R/BaggingWrapper.R

@@ -91,33 +91,38 @@ trainLearner.BaggingWrapper = function(.learner, .task, .subset, .weights = NULL
   .task = subsetTask(.task, subset = .subset)
   n = getTaskSize(.task)
   m = round(n * bw.size)
-  allinds = seq_len(n)
-  if (bw.feats < 1) {
-    feats = getTaskFeatureNames(.task)
-    k = max(round(bw.feats * length(feats)), 1)
-  }
-  models = lapply(seq_len(bw.iters), function(i) {
-    bag = sample(allinds, m, replace = bw.replace)
-    w = .weights[bag]
-    if (bw.feats < 1) {
-      feats2 = sample(feats, k, replace = FALSE)
-      .task2 = subsetTask(.task, features = feats2)
-      train(.learner$next.learner, .task2, subset = bag, weights = w)
-    } else {
-      train(.learner$next.learner, .task, subset = bag, weights = w)
-    }
-  })
-  m = makeHomChainModel(.learner, models)
+  if (bw.feats < 1L)
+    k = max(round(bw.feats * getTaskNFeats(.task)), 1)
+  else
+    k = NULL
+
+  args = list("n" = n, "m" = m, "k" = k, "bw.replace" = bw.replace, "bw.feats" = bw.feats,
+              "task" = .task, "learner" = .learner, "weights" = .weights)
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  models = parallelMap(doBaggingTrainIteration, i = seq_len(bw.iters), more.args = args, level = "mlr.ensemble")
+  makeHomChainModel(.learner, models)
+}
+
+doBaggingTrainIteration = function(i, n, m, k, bw.replace, bw.feats, task, learner, weights) {
+  setSlaveOptions()
+  bag = sample(seq_len(n), m, replace = bw.replace)
+  if (bw.feats < 1L)
+    .task = subsetTask(task, features = sample(getTaskFeatureNames(task), k, replace = FALSE))
+  train(learner$next.learner, task, subset = bag, weights = weights[bag])
 }
 
 #' @export
 predictLearner.BaggingWrapper = function(.learner, .model, .newdata, ...) {
   models = getLearnerModel(.model, more.unwrap = FALSE)
   g = if (.learner$type == "classif") as.character else identity
-  p = asMatrixCols(lapply(models, function(m) {
-    nd = .newdata[, m$features, drop = FALSE]
-    g(predict(m, newdata = nd, ...)$data$response)
-  }))
+
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  p = asMatrixCols(parallelMap(doBaggingPredictIteration, m = models,
+                               more.args = c(list("newdata" = .newdata, "g" = g), list(...)),
+                               level = "mlr.ensemble"))
+
   if (.learner$predict.type == "response") {
     if (.learner$type == "classif")
       as.factor(apply(p, 1L, computeMode))
@@ -137,6 +142,9 @@ predictLearner.BaggingWrapper = function(.learner, .model, .newdata, ...) {
   }
 }
 
+doBaggingPredictIteration = function(m, newdata, g, ...)
+  g(getPredictionResponse(predict(m, newdata = newdata[, m$features, drop = FALSE], ...)))
+
 # we need to override here. while the predtype of the encapsulated learner must always
 # be response, we can estimates probs and se on the outside
 #' @export

R/CostSensRegrWrapper.R

@@ -26,21 +26,23 @@ makeCostSensRegrWrapper = function(learner) {
 trainLearner.CostSensRegrWrapper = function(.learner, .task, .subset, ...) {
   # note that no hyperpars can be in ..., they would refer to the wrapper
   .task = subsetTask(.task, subset = .subset)
-  costs = getTaskCosts(.task)
-  td = getTaskDescription(.task)
-  classes = td$class.levels
-  models = vector("list", length = length(classes))
-  for (i in seq_along(classes)) {
-    cl = classes[i]
-    y = costs[, cl]
-    data = cbind(getTaskData(.task), ..y.. = y)
-    task = makeRegrTask(id = cl, data = data, target = "..y..",
-      check.data = FALSE, fixup.data = "quiet")
-    models[[i]] = train(.learner$next.learner, task)
-  }
+  d = getTaskData(.task)
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  models = parallelMap(doCostSensRegrTrainIteration, cl = getTaskDescription(.task)$class.levels,
+                       more.args = list("d" = d, "costs" = getTaskCosts(.task), "learner" = .learner),
+                       level = "mlr.ensemble")
   makeHomChainModel(.learner, models)
 }
 
+doCostSensRegrTrainIteration = function(learner, cl, costs, d) {
+  setSlaveOptions()
+  data = cbind(d, ..y.. = costs[, cl])
+  task = makeRegrTask(id = cl, data = data, target = "..y..",
+                      check.data = FALSE, fixup.data = "quiet")
+  train(learner$next.learner, task)
+}
+
 #' @export
 predictLearner.CostSensRegrWrapper = function(.learner, .model, .newdata, ...) {
   p = predictHomogeneousEnsemble(.learner, .model, .newdata, ...)

R/HomogeneousEnsemble.R

@@ -49,15 +49,19 @@ getLearnerModel.HomogeneousEnsembleModel = function(model, more.unwrap = FALSE)
 predictHomogeneousEnsemble = function(.learner, .model, .newdata, ...) {
   models = getLearnerModel(.model, more.unwrap = FALSE)
   # for classif we convert factor to char, nicer to handle later on
-  preds = lapply(models, function(mod) {
-    p = predict(mod, newdata = .newdata, ...)$data$response
-    if (is.factor(p))
-      p = as.character(p)
-    return(p)
-  })
-  do.call(cbind, preds)
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  parallelMap(doHomogeneousEnsemblePredictIteration, m = models, simplify = TRUE,
+              more.args = c(list(newdata = .newdata), list(...)), level = "mlr.ensemble")
 }
 
+doHomogeneousEnsemblePredictIteration = function(m, newdata, ...) {
+  setSlaveOptions()
+  p = predict(m, newdata = newdata, ...)$data$response
+  if (is.factor(p))
+    p = as.character(p)
+  return(p)
+}
 
 # call this at end of trainLearner.CostSensRegrWrapper
 # FIXME: potentially remove this when ChainModel is removed

R/MulticlassWrapper.R

@@ -55,20 +55,26 @@ trainLearner.MulticlassWrapper = function(.learner, .task, .subset, .weights = N
   y = getTaskTargets(.task)
   cm = buildCMatrix(mcw.method, .task)
   x = multi.to.binary(y, cm)
-  # now fit models
-  models = lapply(seq_along(x$row.inds), function(i) {
-    data2 = d[x$row.inds[[i]], , drop = FALSE]
-    data2[, tn] = x$targets[[i]]
-    ct = changeData(.task, data2)
-    ct$task.desc$positive = "1"
-    ct$task.desc$negative = "-1"
-    train(.learner$next.learner, ct, weights = .weights)
-  })
+  args = list("x" = x, "d" = getTaskData(.task), "y" = getTaskTargets(.task), "learner" = .learner,
+              "task" = .task, "tn" = getTaskTargetNames(.task), "weights" = .weights)
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  models = parallelMap(i = seq_along(x$row.inds), doMulticlassTrainIteration,
+                       more.args = args, level = "mlr.ensemble")
   m = makeHomChainModel(.learner, models)
   m$cm = cm
   return(m)
 }
 
+doMulticlassTrainIteration = function(x, i, d, y, learner, task, tn, weights) {
+  setSlaveOptions()
+  data2 = d[x$row.inds[[i]],, drop = FALSE]
+  data2[, tn] = x$targets[[i]]
+  ct = changeData(task, data2)
+  ct$task.desc$positive = "1"
+  ct$task.desc$negative = "-1"
+  train(learner$next.learner, ct, weights = weights)
+}
 
 #' @export
 predictLearner.MulticlassWrapper = function(.learner, .model, .newdata, ...) {
@@ -82,6 +88,11 @@ predictLearner.MulticlassWrapper = function(.learner, .model, .newdata, ...) {
       pred = as.numeric(pred == "1") * 2 - 1
     pred
   })
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  p = parallelMap(doMulticlassPredictIteration, m = models,
+                  more.args = c(list("newdata" = .newdata), list(...)), simplify = TRUE,
+                  level = "mlr.ensemble")
   rns = rownames(cm)
   # we use hamming decoding here, see http://jmlr.org/papers/volume11/escalera10a/escalera10a.pdf
   y = apply(p, 1L, function(v) {
@@ -91,6 +102,13 @@ predictLearner.MulticlassWrapper = function(.learner, .model, .newdata, ...) {
   as.factor(y)
 }
 
+doMulticlassPredictIteration = function(m, newdata, ...) {
+  pred = predict(m, newdata = newdata, ...)$data$response
+  if (is.factor(pred))
+    pred = as.numeric(pred == "1") * 2 - 1
+  pred  
+}
+
 #' @export
 getLearnerProperties.MulticlassWrapper = function(learner){
   props = getLearnerProperties(learner$next.learner)

R/MultilabelBinaryRelevanceWrapper.R

@@ -40,21 +40,30 @@ trainLearner.MultilabelBinaryRelevanceWrapper = function(.learner, .task, .subse
   targets = getTaskTargetNames(.task)
   .task = subsetTask(.task, subset = .subset)
   data = getTaskData(.task)
-  models = namedList(targets)
-  for (tn in targets) {
-    data2 = dropNamed(data, setdiff(targets, tn))
-    ctask = makeClassifTask(id = tn, data = data2, target = tn)
-    models[[tn]] = train(.learner$next.learner, ctask, weights = .weights)
-  }
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  models = parallelMap(doMultilabelBinaryRelevanceTrainIteration, tn = targets,
+                       more.args = list(weights = .weights, learner = .learner$next.learner, task = .task,
+                                        data = data), level = "mlr.ensemble")
   makeHomChainModel(.learner, models)
 }
 
 #' @export
 predictLearner.MultilabelBinaryRelevanceWrapper = function(.learner, .model, .newdata, ...) {
   models = getLearnerModel(.model, more.unwrap = FALSE)
-  f = if (.learner$predict.type == "response")
+  f = if (.learner$predict.type == "response") {
     function(m) as.logical(getPredictionResponse(predict(m, newdata = .newdata, ...)))
-  else
-    function(m) getPredictionProbabilities(predict(m, newdata = .newdata, ...), cl = "TRUE")
-  asMatrixCols(lapply(models, f))
+  } else {
+    function(m) getPredictionProbabilities(predict(m, newdata = .newdata, ...))
+  }
+
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  asMatrixCols(parallelMap(f, m = models, level = "mlr.ensemble"), col.names = .model$task.desc$class.levels)
+}
+
+doMultilabelBinaryRelevanceTrainIteration = function(tn, learner, task, data, weights) {
+  setSlaveOptions()
+  task = makeClassifTask(id = tn, data = dropNamed(data, setdiff(getTaskTargetNames(task), tn)), target = tn)
+  train(learner, task, weights = weights)
 }

R/OverBaggingWrapper.R

@@ -83,13 +83,21 @@ trainLearner.OverBaggingWrapper = function(.learner, .task, .subset, .weights =
     z = getMinMaxClass(y)
     obw.cl = z$min.name
   }
-  models = lapply(seq_len(obw.iters), function(i) {
-    bag = sampleBinaryClass(y, rate = obw.rate, cl = obw.cl, resample.other.class = (obw.maxcl == "boot"))
-    train(.learner$next.learner, .task, subset = bag, weights = .weights)
-  })
+  args = list("y" = y, "obw.rate" = obw.rate, "obw.maxcl" = obw.maxcl, "obw.cl" = obw.cl,
+              "learner" = .learner, "task" = .task, "weights" = .weights)
+  parallelLibrary("mlr", master = FALSE, level = "mlr.ensemble", show.info = FALSE)
+  exportMlrOptions(level = "mlr.ensemble")
+  models = parallelMap(doOverBaggingTrainIteration, i = seq_len(obw.iters), more.args = args)
   makeHomChainModel(.learner, models)
 }
 
+doOverBaggingTrainIteration = function(i, y, obw.rate, obw.cl, obw.maxcl, learner, task, weights) {
+  setSlaveOptions()
+  bag = sampleBinaryClass(y, rate = obw.rate, cl = obw.cl, resample.other.class = (obw.maxcl == "boot"))
+  train(learner$next.learner, task, subset = bag, weights = weights)
+}
+
+
 #' @export
 getLearnerProperties.OverBaggingWrapper = function(learner) {
   union(getLearnerProperties(learner$next.learner), "prob")