Apply function to multiple groups using Rcpp and R

Question

I'm trying to apply a function to multiple groups/id's in r using the foreach package. It's taking forever to run using parallel processing via %dopar%, so I was wondering if it's possible to run the apply or for loop portion in c++ via rcpp or other packages to make it faster. I'm not familiar with c++ or other packages that can do this so I'm hoping to learn if this is possible. The sample code is below. My actual function is longer with over 20 inputs and takes even longer to run than what I'm posting

I appreciate the help.

EDIT: I realized my initial question was vague so I'll try to do a better job. I have a table with time series data by group. Each group has > 10K rows. I have written a function in c++ via rcpp that filters the table by group and applies a function. I would like to loop through the unique groups and combine the results like rbind does using rcpp so that it runs faster. See sample code below (my actual function is longer)

library(data.table)
library(inline)
library(Rcpp)
library(stringi)
library(Runuran)

# Fake data
DT <- data.table(Group = rep(do.call(paste0, Map(stri_rand_strings, n=10, length=c(5, 4, 1),
                                                   pattern = c('[A-Z]', '[0-9]', '[A-Z]'))), 180))

df <- DT[order(Group)][
  , .(Month = seq(1, 180, 1),
      Col1 = urnorm(180, mean = 500, sd = 1, lb = 5, ub = 1000), 
      Col2 = urnorm(180, mean = 1000, sd = 1, lb = 5, ub = 1000), 
      Col3 = urnorm(180, mean = 300, sd = 1, lb = 5, ub = 1000)), 
  by = Group
  ]

# Rcpp function
#include <Rcpp.h>
using namespace Rcpp;
// [[Rcpp::plugins(cpp11)]]

// [[Rcpp::export]]
DataFrame testFunc(DataFrame df, StringVector ids, double var1, double var2) {

  // Filter by group
  using namespace std;  
  StringVector sub = df["Group"];
  std::string level = Rcpp::as<std::string>(ids[0]);
  Rcpp::LogicalVector ind(sub.size());
  for (int i = 0; i < sub.size(); i++){
    ind[i] = (sub[i] == level);
  }

  // Access the columns
  CharacterVector Group = df["Group"];
  DoubleVector Month = df["Month"];
  DoubleVector Col1 = df["Col1"];
  DoubleVector Col2 = df["Col2"];
  DoubleVector Col3 = df["Col3"];


  // Create calculations
  DoubleVector Cola = Col1 * (var1 * var2);
  DoubleVector Colb = Col2 * (var1 * var2);
  DoubleVector Colc = Col3 * (var1 * var2);
  DoubleVector Cold = (Cola + Colb + Colc);

  // Result summary
  std::string Group_ID = level;
  double SumCol1 = sum(Col1);
  double SumCol2 = sum(Col2);
  double SumCol3 = sum(Col3);
  double SumColAll = sum(Cold);

  // return a new data frame
  return DataFrame::create(_["Group_ID"]= Group_ID, _["SumCol1"]= SumCol1,
                            _["SumCol2"]= SumCol2, _["SumCol3"]= SumCol3, _["SumColAll"]= SumColAll);
}

# Test function
Rcpp::sourceCpp('sample.cpp')
testFunc(df, ids = "BFTHU1315C", var1 = 24, var2 = 76) # ideally I would like to loop through all groups (unique(df$Group))

#     Group_ID  SumCol1 SumCol2  SumCol3  SumColAll
# 1 BFTHU1315C 899994.6 1798561 540001.6 5907129174

Thanks in advance.

Answer 1

I would suggest to rethink our approach. Your test data set, which I assume is comparable to your real data set, has 3e8 rows. I am estimating about 10 GB of data. You seem to do the following with this data:

• Determine the list of unique IDs (about 5e5)
• Create one task per unique ID
• Each of these tasks gets the full data set and filters out all data that does not belong to the ID in question
• Each of these tasks adds some additional columns that do not depend on the ID
• Each of the tasks does a group_b(ID), but there is only one ID left in the data set
• Each of the tasks calculates some simple means

To me this appears very inefficient w.r.t. memory usage. Generally speaking for problems like this you would want "shared memory parallelism", but foreach gives you only "process parallelism". The downside of process parallelism is that it increases the memory cost.

In addition, you are throwing away all the grouping and aggregation code that exists in base R / dplyr / data.table / SQL engines / ... It is very unlikely that you or any one reading your question here would be able to improve on these existing code bases.

My suggestions:

• Forget about "process parallelism" (for now)
• If you have sufficient RAM, try with a simple dplyr pipe with mutate / group_by / summarize.
• If that is not fast enough, learn how aggregation works with data.table, which is known to be faster and offers "shared memory paralleism" via OpenMP.
• If your computer does not have enough memory and is swapping, then look into possibilities for out-of-memory computation. Personally I would use a (embedded) database.

To make this more explicit. Here a data.table only solution:

library(data.table)
library(stringi)

# Fake data
set.seed(42)
var1 <- 24
var2 <- 76

DT <- data.table(Group = rep(do.call(paste0, Map(stri_rand_strings, n=10, length=c(5, 4, 1),
                                                 pattern = c('[A-Z]', '[0-9]', '[A-Z]'))), 180))
setkey(df, Group)

df <- DT[order(Group)][
  , .(Month = seq(1, 180, 1),
      Col1 = rnorm(180, mean = 500, sd = 1), 
      Col2 = rnorm(180, mean = 1000, sd = 1), 
      Col3 = rnorm(180, mean = 300, sd = 1)), 
  by = Group
  ][, c("Cola", "Colb", "Colc") := .(Col1 * (var1 * var2), 
                                     Col2 * (var1 * var2),
                                     Col3 * (var1 * var2))
    ][, Cold := Cola + Colb + Colc]


# aggregagation
df[, .(SumCol1 = sum(Col1),
       SumCol2 = sum(Col2),
       SumCol3 = sum(Col3),
       SumColAll = sum(Cold)), by = Group]

I am adding the computed columns by reference. The aggregation step uses the grouping functionality provided by data.table. In case your aggregation is more complicated, you can also use a function:

# aggregation function
mySum <- function(Col1, Col2, Col3, Cold) {
  list(SumCol1 = sum(Col1),
       SumCol2 = sum(Col2),
       SumCol3 = sum(Col3),
       SumColAll = sum(Cold))
}

df[, mySum(Col1, Col2, Col3, Cold), by = Group]

And if the aggregation might be faster when using C++ (not the case for things like sum!), you can even use that:

# aggregation function in C++
Rcpp::cppFunction('
Rcpp::List mySum(Rcpp::NumericVector Col1, 
                 Rcpp::NumericVector Col2, 
                 Rcpp::NumericVector Col3, 
                 Rcpp::NumericVector Cold) {
    double SumCol1 = Rcpp::sum(Col1);
    double SumCol2 = Rcpp::sum(Col2);
    double SumCol3 = Rcpp::sum(Col3);
    double SumColAll = Rcpp::sum(Cold);             
    return Rcpp::List::create(Rcpp::Named("SumCol1") = SumCol1,
                              Rcpp::Named("SumCol2") = SumCol2,
                              Rcpp::Named("SumCol3") = SumCol3,
                              Rcpp::Named("SumColAll") = SumColAll);
}
')

df[, mySum(Col1, Col2, Col3, Cold), by = Group]

In all these examples the groping and looping is left to data.table, since you won't gain anything by doing this yourself.