A common scenario is to split a data set into subsets and then apply the same analysis to each part. In context of pipelines, this means that we would like to apply the same pipeline multiple times to each data subset. In addition, we may then want to combine parts of the individual output. As we will see, {pipeflow} provides a built-in function to handle this scenario.
Let’s first define our pipeline, which, to keep matters simple, just fits a linear model and outputs the model coefficients.
library(pipeflow)
pip <- pip_new("linear-model-coeffs") |>
pip_add("data", \(data = NULL) data) |>
pip_add(
"fit",
\(data = ~data, xVar = "x", yVar = "y") {
lm(paste(yVar, "~", xVar), data = data)
}
) |>
pip_add("coefs", \(fit = ~fit) coefficients(fit))So our pipeline looks like this:
pip
# <pipeflow> linear-model-coeffs (3 steps)
# ----------------------------------------
# step params depends state
# 1: data data new
# 2: fit data,xVar,yVar data new
# 3: coefs fit fit new
# ----------------------------------------
# <ready> last run: never
# steps: 3 newOr graphically:
We use the iris data set as our working example.
head(iris)
# Sepal.Length Sepal.Width Petal.Length Petal.Width Species
# 1 5.1 3.5 1.4 0.2 setosa
# 2 4.9 3.0 1.4 0.2 setosa
# 3 4.7 3.2 1.3 0.2 setosa
# 4 4.6 3.1 1.5 0.2 setosa
# 5 5.0 3.6 1.4 0.2 setosa
# 6 5.4 3.9 1.7 0.4 setosaFirst, we apply the pipeline to the whole data set.
pip |> pip_set_params(list(
data = iris,
xVar = "Sepal.Length",
yVar = "Sepal.Width"
))
pip_run(pip)
# info [2026-11-15 18:27:56.172 UTC]: Starting run of pipeflow 'linear-model-coeffs'
# info [2026-11-15 18:27:56.172 UTC]: Step 1/3 data
# info [2026-11-15 18:27:56.173 UTC]: Step 2/3 fit
# info [2026-11-15 18:27:56.174 UTC]: Step 3/3 coefs
# info [2026-11-15 18:27:56.175 UTC]: Finished in 0.003 s (3 done, 0 skipped, 0 failed)Next, we want to apply the pipeline to each species separately. One
way to do this would be to use R’s split function. We can
split it by the Species column and then run the pipeline
for each subset. For example:
run_pipeline_helper <- function(data) {
pip |> pip_set_params(list(data = data))
pip_run(pip)
pip[["coefs", "out"]]
}
results <- lapply(split(iris, iris$Species), FUN = run_pipeline_helper)
# info [2026-11-15 18:27:56.202 UTC]: Starting run of pipeflow 'linear-model-coeffs'
# info [2026-11-15 18:27:56.202 UTC]: Step 1/3 data
# info [2026-11-15 18:27:56.203 UTC]: Step 2/3 fit
# info [2026-11-15 18:27:56.206 UTC]: Step 3/3 coefs
# info [2026-11-15 18:27:56.207 UTC]: Finished in 0.006 s (3 done, 0 skipped, 0 failed)
# info [2026-11-15 18:27:56.210 UTC]: Starting run of pipeflow 'linear-model-coeffs'
# info [2026-11-15 18:27:56.210 UTC]: Step 1/3 data
# info [2026-11-15 18:27:56.211 UTC]: Step 2/3 fit
# info [2026-11-15 18:27:56.212 UTC]: Step 3/3 coefs
# info [2026-11-15 18:27:56.213 UTC]: Finished in 0.003 s (3 done, 0 skipped, 0 failed)
# info [2026-11-15 18:27:56.214 UTC]: Starting run of pipeflow 'linear-model-coeffs'
# info [2026-11-15 18:27:56.214 UTC]: Step 1/3 data
# info [2026-11-15 18:27:56.215 UTC]: Step 2/3 fit
# info [2026-11-15 18:27:56.216 UTC]: Step 3/3 coefs
# info [2026-11-15 18:27:56.216 UTC]: Finished in 0.002 s (3 done, 0 skipped, 0 failed)results
# $setosa
# (Intercept) Sepal.Length
# -0.5694327 0.7985283
#
# $versicolor
# (Intercept) Sepal.Length
# 0.8721460 0.3197193
#
# $virginica
# (Intercept) Sepal.Length
# 1.4463054 0.2318905Unfortunately, with this approach we had to create additional code that had to be run outside the pipeline framework. In addition, the run log quickly can become redundant and confusing, as it now contains multiple runs of the same pipeline. Since splitting data sets (or more generally mapping function calls to different subsets of data) is such a common scenario, {pipeflow} provides a built-in mechanism to handle this case.
Since version 0.3.0, for each step, it is possible to set the
so-called execution mode, which by default is
exec = "auto". To model the above scenario, we add a new
step to our pipeline that splits the data set and set its execution mode
to split.
pip <- pip_new("my-split-pip") |>
pip_add("data", \(data = NULL) data) |>
pip_add(
"split_data",
\(data = ~data, byVar = "by") {
split(data, f = data[[byVar]])
},
exec = "split" # <-- set execution mode to "split"
) |>
pip_add(
"fit",
\(data = ~split_data, xVar = "x", yVar = "y") {
lm(paste(yVar, "~", xVar), data = data)
}
) |>
pip_add("coefs", \(fit = ~fit) coefficients(fit))pip
# <pipeflow> my-split-pip (4 steps)
# ---------------------------------
# step params depends state exec
# 1: data data new auto
# 2: split_data data,byVar data new split
# 3: fit data,xVar,yVar split_data new auto
# 4: coefs fit fit new auto
# ---------------------------------
# <ready> last run: never
# steps: 4 newFirst of all, we see that the pipeline now is printed with an
additional column exec marking the split
execution mode for the split_data step. This also can be
inspected in the graph:
Now what does this execution mode actually do? It basically tells the
pipeline that for all steps that depend on the split_data
step (directly or indirectly), the results coming from the split step
should be treated as lists of results, which should be iterated
over.
In our particular example, this means that the fit step
will be executed for each data subset coming from the
split_data step and likewise the coefs step
will be executed for each fitted model coming from the fit
step.
Let’s see this in action by running the pipeline.
pip |> pip_set_params(list(
data = iris,
xVar = "Sepal.Length",
yVar = "Sepal.Width",
byVar = "Species"
))
pip_run(pip)
# info [2026-11-15 18:27:56.303 UTC]: Starting run of pipeflow 'my-split-pip'
# info [2026-11-15 18:27:56.303 UTC]: Step 1/4 data
# info [2026-11-15 18:27:56.304 UTC]: Step 2/4 split_data
# info [2026-11-15 18:27:56.305 UTC]: Step 3/4 fit
# info [2026-11-15 18:27:56.308 UTC]: Step 4/4 coefs
# info [2026-11-15 18:27:56.309 UTC]: Finished in 0.006 s (4 done, 0 skipped, 0 failed)Looking at the pipeline overview, we see that the
outputs following the split_data steps are now
all lists of results.
pip
# <pipeflow> my-split-pip (4 steps)
# ---------------------------------
# step params depends state out exec
# 1: data data done <data.frame[150x5]> auto
# 2: split_data data,byVar data done <list[3]> split
# 3: fit data,xVar,yVar split_data done <list[3]> auto
# 4: coefs fit fit done <list[3]> auto
# ---------------------------------
# <ready> last run: 2026-11-15 19:27:56 (0.006 s)
# steps: 4 doneInspecting in particular the output of the coefs step,
we see that it is now a list of coefficient tables, one for each
species.
pip[["coefs", "out"]]
# $setosa
# (Intercept) Sepal.Length
# -0.5694327 0.7985283
#
# $versicolor
# (Intercept) Sepal.Length
# 0.8721460 0.3197193
#
# $virginica
# (Intercept) Sepal.Length
# 1.4463054 0.2318905
#
# attr(,"class")
# [1] "list" "pipeflow_partitioned"This matches the output1 we obtained earlier with the helper function but was obtained without the need having to write all this extra code around the pipeline.
While the above approach looks nice already, we are only half way there, because often we will want to recombine the output of all the different subsets in some way. For example, we may want to show the resulting coefficients of the linear models in one summary table.
This is where the reduce execution mode comes into play.
Let’s for this matter extend our pipeline by one step at the end.
pip |> pip_add(
"combine_coefs",
\(coefs = ~coefs) as.data.frame(do.call(rbind, coefs)),
exec = "reduce" # <-- set execution mode to "reduce"
)pip
# <pipeflow> my-split-pip (5 steps)
# ---------------------------------
# step params depends state out exec
# 1: data data done <data.frame[150x5]> auto
# 2: split_data data,byVar data done <list[3]> split
# 3: fit data,xVar,yVar split_data done <list[3]> auto
# 4: coefs fit fit done <list[3]> auto
# 5: combine_coefs coefs coefs new [NULL] reduce
# ---------------------------------
# <ready> last run: 2026-11-15 19:27:56 (0.006 s)
# steps: 4 done, 1 newAgain, we see that the new step is marked with the execution mode
(reduce) in the overview. Graphically, this mode is
represented by a circle.
If we now run the pipeline, we see that the output of the
combine_coefs step is a combined table of coefficients.
pip_run(pip)
# info [2026-11-15 18:27:56.415 UTC]: Starting run of pipeflow 'my-split-pip'
# info [2026-11-15 18:27:56.415 UTC]: Step 1/5 data - skipping done step
# info [2026-11-15 18:27:56.415 UTC]: Step 2/5 split_data - skipping done step
# info [2026-11-15 18:27:56.416 UTC]: Step 3/5 fit - skipping done step
# info [2026-11-15 18:27:56.416 UTC]: Step 4/5 coefs - skipping done step
# info [2026-11-15 18:27:56.416 UTC]: Step 5/5 combine_coefs
# info [2026-11-15 18:27:56.417 UTC]: Finished in 0.002 s (1 done, 4 skipped, 0 failed)
pip[["combine_coefs", "out"]]
# (Intercept) Sepal.Length
# setosa -0.5694327 0.7985283
# versicolor 0.8721460 0.3197193
# virginica 1.4463054 0.2318905If the step function fails for some keys in a run with
pip_run(on_error = "continue"), the step keeps the results
of the other keys, and the steps after it go on with them. See the
vignette on error handling for
the details.
By default, the keys of a mapped step (here fit and
coefs) are run one after another. pip_run()
takes a map function like lapply() that runs
them instead, for example in parallel processes. On Linux and macOS,
parallel::mclapply() does this without any setup:
mc2 <- \(X, FUN) parallel::mclapply(X, FUN, mc.cores = 2)
pip_run(pip, force = TRUE, map = mc2)
# info [2026-11-15 18:27:56.434 UTC]: Starting run of pipeflow 'my-split-pip'
# info [2026-11-15 18:27:56.434 UTC]: Step 1/5 data
# info [2026-11-15 18:27:56.435 UTC]: Step 2/5 split_data
# info [2026-11-15 18:27:56.437 UTC]: Step 3/5 fit
# info [2026-11-15 18:27:56.455 UTC]: Step 4/5 coefs
# info [2026-11-15 18:27:56.470 UTC]: Step 5/5 combine_coefs
# info [2026-11-15 18:27:56.474 UTC]: Finished in 0.040 s (5 done, 0 skipped, 0 failed)
pip[["combine_coefs", "out"]]
# (Intercept) Sepal.Length
# setosa -0.5694327 0.7985283
# versicolor 0.8721460 0.3197193
# virginica 1.4463054 0.2318905map is called as map(X, FUN), so a backend
that needs more arguments is passed as a wrapper like mc2
above. Functions that already have this form, such as
future.apply::future_lapply(), are passed as they are and
also work on Windows:
library(future.apply)
plan(multisession, workers = 3)
pip_run(pip, force = TRUE, map = future_lapply)The split, reduce and other steps that are
not mapped run as before. A few things to keep in mind:
.self (see the vignette on self-modifying pipelines) or
whose function was made by as.function() and keeps its
inner pipelines (see the next
vignette) are therefore always run one after another.future_lapply() does and
mclapply() doesn’t.future_lapply(..., future.seed = TRUE).See section ‘Mapping in parallel’ of ?pip_run for the
details.
The split and reduce execution modes cover
the common case where the pipeline itself performs the split. Whenever
they fit, they are the simplest way to express such a workflow. If you
instead need to map over inputs that are not a single split point (for
example cross-validation folds or nested splits) or want to reuse a
standalone pipeline as a self-contained building block, see the next vignette on nested
pipelines.
Technically, the output is slightly different, because the returned list has an additional class attribute “pipeflow_partitioned”.↩︎