Run a pipeline — run_pipeline • pippy

run_pipeline() manages the running of a user defined pipeline across multiple scenarios. Different stages of analysis are defined as functions which are then brought together as a pipeline using a syntax that is a reduced subset of R. This pipeline is then represented as a directed acyclic graph allowing us to employ a robust approach to work planning and caching.

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,
  optimise_for_memory = FALSE
)

Arguments

x

[expression]

R Expression of pipeline assignments.

See details for information on the supported syntax.

...

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.

See details for information on the configuration file format.

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.

optimise_for_memory

[bool]

If TRUE then when a target is no longer needed in the calculation of subsequent targets they are removed from memory and a forced gc is called.

When the objects are needed by subsequent scenarios they will be reloaded from disk.

This can only be TRUE when return = FALSE.

Value

A named list of outputs for each configuration.

Details

Normally a pipeline expression will involve multiple assignments need embracing to represent a single expression. The syntax has a few additional requirements:

The left-hand-side (target) of each assignment must be uniquely named.
Function calls on the right-hand-side (rhs) must be defined in the function directory (see relative_function_dir) or contained in the base R or stats namespaces.
The target names cannot match the names of functions in base R, or those contained in the function directory.
The order of assignments is not important as it is determined at runtime.
Values from the configuration file (see relative_config_file) are accessed by the special CONFIG value (a list which can be subset using the $ symbol).
For serialising objects to disk we utilise R's rds format (with the ability to pass arguments through via saveRDS_args). This means that functions seen on the RHS of the expressions should only return objects that are unserialisable by readRDS().

Configuration files are simply R scripts where scenarios are defined by named lists. They are are evaluated in a dedicated environment where we enable two special functions:

INFILE() marks a file as an input. This ensures that run_pipeline() knows to keep a hash of the object and to monitor it for future changes when the pipeline is rerun.
OUTFILE() marks a file name as an output. This ensures that outputs from different scenarios (see later) do not overwrite one another (output files get saved in a scenario-dependent directory). It also ensures that output files are recreated if they are deleted.

Caveats

When working out which stages of a pipeline need rerunning we take in to account functions defined by the user in the function directory. We do not account for functions of called packages or base R and stats. This means that updating packages (or R itself) can potentially invalidate the cache so it is the users responsibility to maintain a consistent environment, or force a rerun of the pipeline where necessary. Future versions of pippy will warn when underlying package versions (or R itself) change between runs.
The only global object we account for is .Random.seed. If you wish to take advantage of object caching then it is important to set a seed before running a pipeline. This seed will then be utilised at the start of each scenario run.

Examples

# generate a demo pipeline with a single scenario
dir <- create_skeleton_pipeline(tempfile())
#> ✔ Template pipeline created in /tmp/RtmpM3nsmh/file5a36508ebf01.

# 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/RtmpM3nsmh/file5a36508ebf01/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 ───────────────────────────
#> ℹ Using object `raw` from memory.
#> ℹ 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)