run_pipeline() manages the running of a user defined pipeline simplifying
object caching and configuration management across different scenarios.
Usage
run_pipeline(
x,
...,
scenario = getOption("pipeline.scenario_default", "default"),
scenario_default = getOption("pipeline.scenario_default", "default"),
pipeline_dir = ".",
relative_config_file = getOption("pipeline.config_file", "config.R"),
relative_function_dir = getOption("pipeline.function_dir", "R"),
relative_output_dir = getOption("pipeline.output_dir", "output"),
force = FALSE,
saveRDS_args = list(),
readRDS_args = list(),
return = TRUE,
gc = FALSE
)Arguments
- x
[expression]R Expression of pipeline assignments. Normally this will involve multiple assignments and will need to be embraced to represent a single expression.
- ...
Not currently used.
- scenario
[character]Scenario(s) to run on top of
scenario_default.These represent the file configurations you wish to loop over. In this case the configuration file will first be parsed looking for a named-list entry corresponding to the value of the
scenario_defaultargument. The chosen scenarios are then layered on top of this default usingutils::modifyList(). The default scenario is always run first.- scenario_default
[character]The default scenario to consider. That is, the scenario on which the specified scenarios are layered.
If no configuration file exists this will represent the folder under the
output_dirwhere file output and cached objects are stored as well as the element of the resulting list output.- pipeline_dir
[character]The directory you wish to run the the pipeline relative to.
- relative_config_file
[character]The configuration file.
Must be a none-nested and given relative to
pipeline_dir.- relative_function_dir
[character]The directory to look for user defined pipeline functions.
Must be a none-nested and given relative to
pipeline_dir.- relative_output_dir
[character]The directory for output.
Must be a none-nested and given relative to
pipeline_dir.- force
[bool]Do you want to force a run of the pipeline.
If TRUE, then cached objects are removed and the pipeline is (re)run.
- saveRDS_args
[list]List of additional arguments passed to
saveRDS().This argument allows you to pass additional arguments to that function (e.g.
list(compress = "zstd"))- readRDS_args
[list]List of additional arguments passed to
readRDS().- return
[bool]Should the output be returned.
Defaults to TRUE, but when running across multiple scenarios and with outputs that use a large amount of memory it can be useful to set
FALSE.If
FALSE, a successful run will returnNULLinvisibly.- gc
[bool]Should we force calls to
gc()whilst running the pipeline.For pipeline's creating large objects, setting this to TRUE may help reduce memory consumption.
Examples
# generate a demo pipeline with a single scenario
dir <- create_skeleton_pipeline(tempfile())
#> ✔ Template pipeline created in /tmp/RtmpiVYZyj/file2a87c259c8f.
# Note the configuration file and the folder of R functions
list.files(dir, all.files = TRUE, recursive = TRUE, no.. = TRUE)
#> [1] "R/load_dat.R" "R/plot_dat.R" "R/wrangle_dat.R" "config.R"
#> [5] "data/mtcars.csv" "pipeline.R"
# Run the pipeline with the default configuration (see config.R)
out <- run_pipeline(
{
raw <- load_dat(CONFIG$in_csv)
clean <- wrangle_dat(raw, CONFIG$rows)
plot <- plot_dat(clean, CONFIG$out_plot)
},
pipeline_dir = dir,
scenario = "default"
)
#>
#> ── Starting pipeline with `default` configuration ──────────────────────────────
#> ℹ Running function to generate object `raw`.
#> ℹ Saving `raw` to disk.
#> ℹ Running function to generate object `clean`.
#> ℹ Saving `clean` to disk.
#> ℹ Running function to generate object `plot`.
#> ℹ Saving `plot` to disk.
#>
#> ── pipeline(s) finished ────────────────────────────────────────────────────────
# output in a list
str(out$default$raw)
#> 'data.frame': 32 obs. of 12 variables:
#> $ car : chr "Mazda RX4" "Mazda RX4 Wag" "Datsun 710" "Hornet 4 Drive" ...
#> $ mpg : num 21 21 22.8 21.4 18.7 18.1 14.3 24.4 22.8 19.2 ...
#> $ cyl : int 6 6 4 6 8 6 8 4 4 6 ...
#> $ disp: num 160 160 108 258 360 ...
#> $ hp : int 110 110 93 110 175 105 245 62 95 123 ...
#> $ drat: num 3.9 3.9 3.85 3.08 3.15 2.76 3.21 3.69 3.92 3.92 ...
#> $ wt : num 2.62 2.88 2.32 3.21 3.44 ...
#> $ qsec: num 16.5 17 18.6 19.4 17 ...
#> $ vs : int 0 0 1 1 0 1 0 1 1 1 ...
#> $ am : int 1 1 1 0 0 0 0 0 0 0 ...
#> $ gear: int 4 4 4 3 3 3 3 4 4 4 ...
#> $ carb: int 4 4 1 1 2 1 4 2 2 4 ...
str(out$default$clean)
#> 'data.frame': 10 obs. of 12 variables:
#> $ car : chr "Mazda RX4" "Mazda RX4 Wag" "Datsun 710" "Hornet 4 Drive" ...
#> $ mpg : num 21 21 22.8 21.4 18.7 18.1 14.3 24.4 22.8 19.2
#> $ cyl : int 6 6 4 6 8 6 8 4 4 6
#> $ disp: num 160 160 108 258 360 ...
#> $ hp : int 110 110 93 110 175 105 245 62 95 123
#> $ drat: num 3.9 3.9 3.85 3.08 3.15 2.76 3.21 3.69 3.92 3.92
#> $ wt : num 2.62 2.88 2.32 3.21 3.44 ...
#> $ qsec: num 16.5 17 18.6 19.4 17 ...
#> $ vs : int 0 0 1 1 0 1 0 1 1 1
#> $ am : int 1 1 1 0 0 0 0 0 0 0
#> $ gear: int 4 4 4 3 3 3 3 4 4 4
#> $ carb: int 4 4 1 1 2 1 4 2 2 4
out$default$plot
#> [1] "/tmp/RtmpiVYZyj/file2a87c259c8f/output/default/plot.png"
#> attr(,"OUTFILE")
#> [1] TRUE
# run with the 'production' configuration as well (see config.R)
out <- run_pipeline(
{
raw <- load_dat(CONFIG$in_csv)
clean <- wrangle_dat(raw, CONFIG$rows)
plot <- plot_dat(clean, CONFIG$out_plot)
},
pipeline_dir = dir,
scenario = c("default", "production")
)
#>
#> ── Starting pipeline with `default` configuration ──────────────────────────────
#> ℹ Loading object `raw` from disk cache.
#> ℹ Loading object `clean` from disk cache.
#> ℹ Loading object `plot` from disk cache.
#>
#> ── Starting pipeline with `production` configuration ───────────────────────────
#> ℹ Loading object `raw` from disk cache.
#> ℹ Running function to generate object `clean`.
#> ℹ Saving `clean` to disk.
#> ℹ Running function to generate object `plot`.
#> ℹ Saving `plot` to disk.
#>
#> ── pipeline(s) finished ────────────────────────────────────────────────────────
unlink(dir)