<p><div><div><div><div data-show-faces="false" data-height="21" data-width="100" data-layout="button_count" data-send="true" data-href="https://www.r-bloggers.com/speed-up-your-r-work/"></div></div><div></div></div><div></div></div>
<div>
(This article was first published on <strong><a href="http://www.win-vector.com/blog/2018/07/speed-up-your-r-work/"> R – Win-Vector Blog</a></strong>, and kindly contributed to <a rel="nofollow" href="https://www.r-bloggers.com/">R-bloggers)</a> </div>
<h1>Introduction</h1>
<p>In this note we will show how to speed up work in <a rel="nofollow" href="https://www.r-project.org/" target="_blank"><code>R</code></a> by partitioning data and process-level parallelization. We will show the technique with three different <code>R</code> packages: <a rel="nofollow" href="https://github.com/WinVector/rqdatatable" target="_blank"><code>rqdatatable</code></a>, <a rel="nofollow" href="https://cran.r-project.org/package=data.table" target="_blank"><code>data.table</code></a>, and <a rel="nofollow" href="https://cran.r-project.org/package=dplyr" target="_blank"><code>dplyr</code></a>. The methods shown will also work with base-<code>R</code> and other packages.</p>
<p>For each of the above packages we speed up work by using <a rel="nofollow" href="https://winvector.github.io/wrapr/reference/execute_parallel.html" target="_blank"><code>wrapr::execute_parallel</code></a> which in turn uses <a rel="nofollow" href="https://winvector.github.io/wrapr/reference/partition_tables.html" target="_blank"><code>wrapr::partition_tables</code></a> to partition un-related <code>data.frame</code> rows and then distributes them to different processors to be executed. <a rel="nofollow" href="https://winvector.github.io/rqdatatable/reference/ex_data_table_parallel.html" target="_blank"><code>rqdatatable::ex_data_table_parallel</code></a> conveniently bundles all of these steps together when working with <a rel="nofollow" href="https://cran.r-project.org/package=rquery" target="_blank"><code>rquery</code></a> pipelines.</p>
<p>The partitioning is specified by the user preparing a grouping column that tells the system which sets of rows must be kept together in a correct calculation. We are going to try to demonstrate everything with simple code examples, and minimal discussion.</p>
<p><span></span></p>
<p>
<p>Keep in mind: unless the pipeline steps have non-trivial cost, the overhead of partitioning and distributing the work may overwhelm any parallel speedup. Also <code>data.table</code> itself already seems to exploit some thread-level parallelism (notice user time is greater than elapsed time). That being said, in this note we will demonstrate a synthetic example where computation is expensive due to a blow-up in an intermediate join step.</p>
<h1>Our example</h1>
<p>First we set up our execution environment and example (some details: OSX 10.13.4 on a 2.8 GHz Intel Core i5 Mac Mini (Late 2015 model) with 8GB RAM and hybrid disk drive).</p>
<div>
<pre><code><span>library</span>(<span>&quot;rqdatatable&quot;</span>)</code></pre>
</div>
<pre><code>## Loading required package: rquery</code></pre>
<div>
<pre><code><span>library</span>(<span>&quot;microbenchmark&quot;</span>)
<span>library</span>(<span>&quot;ggplot2&quot;</span>)
<span>library</span>(<span>&quot;WVPlots&quot;</span>)
<span>suppressPackageStartupMessages</span>(<span>library</span>(<span>&quot;dplyr&quot;</span>))</code></pre>
</div>
<pre><code>## Warning: package 'dplyr' was built under R version 3.5.1</code></pre>
<div>
<pre><code>base::<span>date</span>()</code></pre>
</div>
<pre><code>## [1] &quot;Sun Jul  8 09:05:25 2018&quot;</code></pre>
<div>
<pre><code>R.version.string</code></pre>
</div>
<pre><code>## [1] &quot;R version 3.5.0 (2018-04-23)&quot;</code></pre>
<div>
<pre><code>parallel::<span>detectCores</span>()</code></pre>
</div>
<pre><code>## [1] 4</code></pre>
<div>
<pre><code><span>packageVersion</span>(<span>&quot;parallel&quot;</span>)</code></pre>
</div>
<pre><code>## [1] '3.5.0'</code></pre>
<div>
<pre><code><span>packageVersion</span>(<span>&quot;rqdatatable&quot;</span>)</code></pre>
</div>
<pre><code>## [1] '0.1.2'</code></pre>
<div>
<pre><code><span>packageVersion</span>(<span>&quot;rquery&quot;</span>)</code></pre>
</div>
<pre><code>## [1] '0.5.1'</code></pre>
<div>
<pre><code><span>packageVersion</span>(<span>&quot;dplyr&quot;</span>)</code></pre>
</div>
<pre><code>## [1] '0.7.6'</code></pre>
<div>
<pre><code>ncore &lt;-<span> </span>parallel::<span>detectCores</span>()
<span>print</span>(ncore)</code></pre>
</div>
<pre><code>## [1] 4</code></pre>
<div>
<pre><code>cl &lt;-<span> </span>parallel::<span>makeCluster</span>(ncore)
<span>print</span>(cl)</code></pre>
</div>
<pre><code>## socket cluster with 4 nodes on host 'localhost'</code></pre>
<div>
<pre><code><span>set.seed</span>(<span>2362</span>)
mk_example &lt;-<span> </span>function(nkey, nrep, <span>ngroup =</span> <span>20</span>) {
  keys &lt;-<span> </span><span>paste0</span>(<span>&quot;key_&quot;</span>, <span>seq_len</span>(nkey))
  key_group &lt;-<span> </span><span>sample</span>(<span>as.character</span>(<span>seq_len</span>(ngroup)), 
                      <span>length</span>(keys), <span>replace =</span> <span>TRUE</span>)
  <span>names</span>(key_group) &lt;-<span> </span>keys
  key_table &lt;-<span> </span><span>data.frame</span>(
    <span>key =</span> <span>rep</span>(keys, nrep),
    <span>stringsAsFactors =</span> <span>FALSE</span>)
  key_table$data &lt;-<span> </span><span>runif</span>(<span>nrow</span>(key_table))
  instance_table &lt;-<span> </span><span>data.frame</span>(
    <span>key =</span> <span>rep</span>(keys, nrep),
    <span>stringsAsFactors =</span> <span>FALSE</span>)
  instance_table$id &lt;-<span> </span><span>seq_len</span>(<span>nrow</span>(instance_table))
  instance_table$info &lt;-<span> </span><span>runif</span>(<span>nrow</span>(instance_table))
  <span># groups should be no finer than keys</span>
  key_table$key_group &lt;-<span> </span>key_group[key_table$key]
  instance_table$key_group &lt;-<span> </span>key_group[instance_table$key]
  <span>list</span>(<span>key_table =</span> key_table,
       <span>instance_table =</span> instance_table)
}

dlist &lt;-<span> </span><span>mk_example</span>(<span>10</span>, <span>10</span>)
data &lt;-<span> </span>dlist$instance_table
annotation &lt;-<span> </span>dlist$key_table</code></pre>
</div>
<h1>rquery / rqdatatable</h1>
<p><a rel="nofollow" href="https://github.com/WinVector/rquery" target="_blank"><code>rquery</code></a> and <a rel="nofollow" href="https://github.com/WinVector/rqdatatable" target="_blank"><code>rqdatatable</code></a> can implement a non-trivial calculation as follows.</p>
<div>
<pre><code><span># possible data lookup: find rows that</span>
<span># have lookup data &lt;=info</span>
optree &lt;-<span> </span><span>local_td</span>(data) %.>%
<span> </span><span>natural_join</span>(., 
               <span>local_td</span>(annotation), 
               <span>jointype =</span> <span>&quot;INNER&quot;</span>, 
               <span>by =</span> <span>&quot;key&quot;</span>) %.>%
<span> </span><span>select_rows_nse</span>(., data &lt;=<span> </span>info) %.>%
<span> </span><span>pick_top_k</span>(., 
             <span>k =</span> <span>1</span>,
             <span>partitionby =</span> <span>&quot;id&quot;</span>,
             <span>orderby =</span> <span>&quot;data&quot;</span>,
             <span>reverse =</span> <span>&quot;data&quot;</span>,
             <span>keep_order_column =</span> <span>FALSE</span>) %.>%
<span> </span><span>orderby</span>(., <span>&quot;id&quot;</span>)
<span>cat</span>(<span>format</span>(optree))</code></pre>
</div>
<pre><code>## table('data'; 
##   key,
##   id,
##   info,
##   key_group) %.>%
##  natural_join(.,
##   table('annotation'; 
##     key,
##     data,
##     key_group),
##   j= INNER, by= key) %.>%
##  select_rows(.,
##    data &lt;= info) %.>%
##  extend(.,
##   row_number := row_number(),
##   p= id,
##   o= &quot;data&quot; DESC) %.>%
##  select_rows(.,
##    row_number &lt;= 1) %.>%
##  drop_columns(.,
##    row_number) %.>%
##  orderby(., id)</code></pre>
<div>
<pre><code>res1 &lt;-<span> </span><span>ex_data_table</span>(optree)
<span>head</span>(res1)</code></pre>
</div>
<pre><code>##         data id      info   key key_group
## 1: 0.9152014  1 0.9860654 key_1        20
## 2: 0.5599810  2 0.5857570 key_2         8
## 3: 0.3011882  3 0.3334490 key_3        10
## 4: 0.3650987  4 0.3960980 key_4         5
## 5: 0.1469254  5 0.1753649 key_5        14
## 6: 0.2567631  6 0.3510280 key_6         7</code></pre>
<div>
<pre><code><span>nrow</span>(res1)</code></pre>
</div>
<pre><code>## [1] 94</code></pre>
<p>And we can execute the operations in parallel.</p>
<div>
<pre><code>parallel::<span>clusterEvalQ</span>(cl, 
                       <span>library</span>(<span>&quot;rqdatatable&quot;</span>))</code></pre>
</div>
<pre><code>## [[1]]
## [1] &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;    &quot;grDevices&quot;  
## [6] &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;       
## 
## [[2]]
## [1] &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;    &quot;grDevices&quot;  
## [6] &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;       
## 
## [[3]]
## [1] &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;    &quot;grDevices&quot;  
## [6] &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;       
## 
## [[4]]
## [1] &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;    &quot;grDevices&quot;  
## [6] &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;</code></pre>
<div>
<pre><code>res2 &lt;-<span> </span><span>ex_data_table_parallel</span>(optree, 
                               <span>&quot;key_group&quot;</span>, 
                               cl)
<span>head</span>(res2)</code></pre>
</div>
<pre><code>##         data id      info   key key_group
## 1: 0.9152014  1 0.9860654 key_1        20
## 2: 0.5599810  2 0.5857570 key_2         8
## 3: 0.3011882  3 0.3334490 key_3        10
## 4: 0.3650987  4 0.3960980 key_4         5
## 5: 0.1469254  5 0.1753649 key_5        14
## 6: 0.2567631  6 0.3510280 key_6         7</code></pre>
<div>
<pre><code><span>nrow</span>(res2)</code></pre>
</div>
<pre><code>## [1] 94</code></pre>
<h1>data.table</h1>
<p><a rel="nofollow" href="http://r-datatable.com/" target="_blank"><code>data.table</code></a> can implement the same function.</p>
<div>
<pre><code><span>library</span>(<span>&quot;data.table&quot;</span>)</code></pre>
</div>
<pre><code>## 
## Attaching package: 'data.table'

## The following objects are masked from 'package:dplyr':
## 
##     between, first, last</code></pre>
<div>
<pre><code><span>packageVersion</span>(<span>&quot;data.table&quot;</span>)</code></pre>
</div>
<pre><code>## [1] '1.11.4'</code></pre>
<div>
<pre><code>data_table_f &lt;-<span> </span>function(data, annotation) {
  data &lt;-<span> </span>data.table::<span>as.data.table</span>(data)
  annotation &lt;-<span> </span>data.table::<span>as.data.table</span>(annotation)
  joined &lt;-<span> </span><span>merge</span>(data, annotation, 
                  <span>by =</span> <span>&quot;key&quot;</span>, 
                  <span>all=</span><span>FALSE</span>, 
                  <span>allow.cartesian=</span><span>TRUE</span>)
  joined &lt;-<span> </span>joined[joined$data &lt;=<span> </span>joined$info, ]
  data.table::<span>setorderv</span>(joined, <span>cols =</span> <span>&quot;data&quot;</span>)
  joined &lt;-<span> </span>joined[, .SD[.N], id]
  data.table::<span>setorderv</span>(joined, <span>cols =</span> <span>&quot;id&quot;</span>)
}
resdt &lt;-<span> </span><span>data_table_f</span>(data, annotation)
<span>head</span>(resdt)</code></pre>
</div>
<pre><code>##    id   key      info key_group.x      data key_group.y
## 1:  1 key_1 0.9860654          20 0.9152014          20
## 2:  2 key_2 0.5857570           8 0.5599810           8
## 3:  3 key_3 0.3334490          10 0.3011882          10
## 4:  4 key_4 0.3960980           5 0.3650987           5
## 5:  5 key_5 0.1753649          14 0.1469254          14
## 6:  6 key_6 0.3510280           7 0.2567631           7</code></pre>
<div>
<pre><code><span>nrow</span>(resdt)</code></pre>
</div>
<pre><code>## [1] 94</code></pre>
<p>We can also run <code>data.table</code> in parallel using <a rel="nofollow" href="https://winvector.github.io/wrapr/reference/execute_parallel.html" target="_blank"><code>wrapr::execute_parallel</code></a>.</p>
<div>
<pre><code>parallel::<span>clusterEvalQ</span>(cl, <span>library</span>(<span>&quot;data.table&quot;</span>))</code></pre>
</div>
<pre><code>## [[1]]
##  [1] &quot;data.table&quot;  &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;   
##  [6] &quot;grDevices&quot;   &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;       
## 
## [[2]]
##  [1] &quot;data.table&quot;  &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;   
##  [6] &quot;grDevices&quot;   &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;       
## 
## [[3]]
##  [1] &quot;data.table&quot;  &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;   
##  [6] &quot;grDevices&quot;   &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;       
## 
## [[4]]
##  [1] &quot;data.table&quot;  &quot;rqdatatable&quot; &quot;rquery&quot;      &quot;stats&quot;       &quot;graphics&quot;   
##  [6] &quot;grDevices&quot;   &quot;utils&quot;       &quot;datasets&quot;    &quot;methods&quot;     &quot;base&quot;</code></pre>
<div>
<pre><code>parallel::<span>clusterExport</span>(cl, <span>&quot;data_table_f&quot;</span>)

dt_f &lt;-<span> </span>function(tables_list) {
  data &lt;-<span> </span>tables_list$data
  annotation &lt;-<span> </span>tables_list$annotation
  <span>data_table_f</span>(data, annotation)
}

data_table_parallel_f &lt;-<span> </span>function(data, annotation) {
  respdt &lt;-<span> </span>wrapr::<span>execute_parallel</span>(
    <span>tables =</span> <span>list</span>(<span>data =</span> data, 
                  <span>annotation =</span> annotation),
    <span>f =</span> dt_f,
    <span>partition_column =</span> <span>&quot;key_group&quot;</span>,
    <span>cl =</span> cl) %.>%
<span> </span>data.table::<span>rbindlist</span>(.)
  data.table::<span>setorderv</span>(respdt, <span>cols =</span> <span>&quot;id&quot;</span>)
  respdt
}
respdt &lt;-<span> </span><span>data_table_parallel_f</span>(data, annotation)
<span>head</span>(respdt)</code></pre>
</div>
<pre><code>##    id   key      info key_group.x      data key_group.y
## 1:  1 key_1 0.9860654          20 0.9152014          20
## 2:  2 key_2 0.5857570           8 0.5599810           8
## 3:  3 key_3 0.3334490          10 0.3011882          10
## 4:  4 key_4 0.3960980           5 0.3650987           5
## 5:  5 key_5 0.1753649          14 0.1469254          14
## 6:  6 key_6 0.3510280           7 0.2567631           7</code></pre>
<div>
<pre><code><span>nrow</span>(respdt)</code></pre>
</div>
<pre><code>## [1] 94</code></pre>
<h1>dplyr</h1>
<p><a rel="nofollow" href="https://cran.r-project.org/package=dplyr" target="_blank"><code>dplyr</code></a> can also implement the example.</p>
<div>
<pre><code>dplyr_pipeline &lt;-<span> </span>function(data, annotation) {
  res &lt;-<span> </span>data %>%
<span> </span><span>inner_join</span>(annotation, <span>by =</span> <span>&quot;key&quot;</span>) %>%
<span> </span><span>filter</span>(data &lt;=<span> </span>info) %>%
<span> </span><span>group_by</span>(id) %>%
<span> </span><span>arrange</span>(-data) %>%
<span> </span><span>mutate</span>(<span>rownum =</span> <span>row_number</span>()) %>%
<span> </span><span>ungroup</span>() %>%
<span> </span><span>filter</span>(rownum ==<span> </span><span>1</span>) %>%
<span> </span><span>arrange</span>(id)
  res
}

resd &lt;-<span> </span><span>dplyr_pipeline</span>(data, annotation)
<span>head</span>(resd)</code></pre>
</div>
<pre><code>## # A tibble: 6 x 7
##   key      id  info key_group.x  data key_group.y rownum
##   </code></pre>