Skip to contents

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_default argument. The chosen scenarios are then layered on top of this default using utils::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_dir where 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 return NULL invisibly.

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.

Value

A named list of outputs for each configuration.

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)