---
title: "Introduction to pipeline"
output:
    html:
        meta:
            css: ["@default@1.14.24", "@callout@1.14.24", "@copy-button@1.14.24", "@article@1.14.24", "assets/tweaks.css"]
            js: ["@callout@1.14.24", "@copy-button@1.14.24", "@toc-highlight@1.14.24", "@sidenotes@1.14.24"]
            lang: en
        options:
            toc: true
            js_highlight:
                package: prism
                version: 1.29.0

vignette: >
  %\VignetteEngine{litedown::vignette}
  %\VignetteIndexEntry{Introduction to pipeline}
  %\VignetteEncoding{UTF-8}
---

::: {.callout-warning data-legend="Warning"}
This package:

- is a work-in-progress, proof-of-concept;
- is unstable in both it's API and output; 
- may not fully fledge past the prototype stage.
:::

## Background

The pipeline package's aim is to make it easy for users to adopt a functional approach to their analysis.
Pipelines are defined using a reduced subset of R and, when run, use an intelligent approach to work planning and object caching.
It was motivated by a desire to reduce the boilerplate I was writing with my R scripts and associated Makefiles.
Whilst the goal is to be "good enough" for many pieces of analysis, by encouraging a functional approach, it remains easy to move to more advanced alternatives such as [targets](https://docs.ropensci.org/targets/).


## Scope

Goals:

- 🗸 Intelligent caching to reduce unnecessary execution.
- 🗸 Configuration files using a reduced subset of base R as their syntax.
- 🗸 Simple and efficient approach to managing multiple scenarios.

Out of scope:

- 🗴 HPC orchestration.
- 🗴 Automatic parallelisation of the pipeline. 

## A motivating example

The easiest way to get started with the package is to look at some of the included examples.
`pipeline_skeleton()` can be used to set up a pipeline based on the mtcars data set and across single, or multiple, scenarios.
We start with the single scenario example.

```{r}
#| eval: false
library(pipeline)
dir <- pipeline_skeleton(dir = tempfile(), single = TRUE)
```

```{r}
#| echo: false
library(pipeline)
f <- tempfile()
dir <- tryCatch(
    pipeline_skeleton(f, single = TRUE),
    error = function(cond) {
        unlink(f)
        stop("Unable to create pipeline skeleton.")
    }
)
```

We now have the following in `dir`:

```{r}
#| echo: false
if (requireNamespace("fs", quietly = TRUE)) {
    fs::dir_tree(dir)
} else {
    list.files(dir, all.files = TRUE, recursive = TRUE, no.. = TRUE)
}
```

The pipeline we are going to run is described in `pipeline.R`:

```{.r}
out <- pipeline_run({
    raw   <- load_dat(CONFIG$in_csv)
    clean <- wrangle_dat(raw, CONFIG$rows)
    plot  <- plot_dat(clean, CONFIG$out_plot)
})
```

`pipeline_run()` takes an expression where each assignment it contains represents a stage of the analysis.
There are a few things to note about the expression syntax:

- 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 `R`, contained in the base R namespace or, if present in another package, be explicitly namespaced (i.e. `package::fun`).
- 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 internally by the function.
- Values from the configuration file, `config.R`, are accessed by the special `CONFIG` value (a list which can be subset using the `$` symbol).


### The configuration file

Configuration files are simply R scripts that are evaluated in a dedicated environment (with some allowances for multiple scenarios that we discuss at the end).

Within this environment we enable two special functions `INFILE` and `OUTFILE`:

- `INFILE()` marks a file as an input.
This ensures that `pipeline_run()` 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.

In our example we have the following configuration file:

::: {.callout-example data-legend="config.R"}
```{.r}
in_csv   <- INFILE("data/mtcars.csv")
out_plot <-  OUTFILE("plot.png")
rows     <-  10
```
:::

Here `in_csv` points to our input data and `out_plot` to a png output file.

### The R directory

This is the directory where users define functions that represent stages of their analysis (e.g. cleaning data, modelling, plotting, ...).
Here we have:

::: {.callout-example data-legend="R/load_dat.R"}
```{.r}
load_dat <- function(file) {
    # Note: If function not in the base/stats namespace or function directory
    #       we must explictly namespace the call using `::`.
    utils::read.csv(file)
}
```
:::

This file contains a single function which takes the filename of raw data as input before doing a little wrangling and returning the output.
Note the comment about namespacing functions;
anything not in the base/stats namespace or the function directory needs to be explicitly namespaced.

Next we have `R/wrangle_dat.R`:

::: {.callout-example data-legend="R/wrangle_dat.R"}
```{.r}
wrangle_dat <- function(dat, rows) {
    stopifnot(is.data.frame(dat), nrow(dat) >= rows, rows >= 0)
    head(dat, rows)
}
```
:::

This function curtails the number of rows of data.
Here we also make use of assertions in the function,
checking that `dat` is a data frame and `rows` is between 0 and the number of rows in the data frame (inclusive).
Assertions like these can be thought of unit tests for your analysis. 
Whilst here we have them at the start of the function to check inputs,
they can be used anywhere in your function to ensure things are running as expected.
If an assertion fails it will stop the pipeline running, bringing your attention to any issue at hand.

Finally there is

::: {.callout-example data-legend="R/plot_dat.R"}
```{.r}
plot_dat <- function(dat, outfile) {
    grDevices::png(outfile)
    plot(factor(dat$cyl), dat$mpg, xlab = "cyl", ylab = "mpg")
    grDevices::dev.off()
    # return name of where image is saved
    outfile
}

```
:::

This function takes the manipulated data as input and creates a plot which is saved as a png.
When saving output to disk, we recommend that the function returns the location where the file is saved for later use.

### Putting it together

Running the pipeline we see the following

```{r}
#| eval: false
original_directory <- setwd(dir)
out <- pipeline_run({
    raw   <- load_dat(CONFIG$in_csv)
    clean <- wrangle_dat(raw, CONFIG$rows)
    plot  <- plot_dat(clean, CONFIG$out_plot)
})
```

 

```{r}
#| echo: false
out <- pipeline_run(
    {
        raw   <- load_dat(CONFIG$in_csv)
        clean <- wrangle_dat(raw, CONFIG$rows)
        plot  <- plot_dat(clean, CONFIG$out_plot)
    },
    pipeline_dir = dir
)
```

Each target output is saved within a single list that we have allocated to out.
The name of the list is configurable but, by default, is 'default'.

```{r}
str(out$default$raw)
str(out$default$clean)
out$default$plot
```

```{r}
#| echo: false
#| fig.cap: "The resulting plot in the 'output' directory."
#| fig.alt: "Box plot of cyl (x-axis) versus mpg (y-axis). There is a downward trend showing in mpg as cyl increases."
plot(factor(out$default$clean$cyl), out$default$clean$mpg, xlab = "cyl", ylab = "mpg")
```

The outputs, along with some additional information that we can use for cache invalidation checks, are also saved to disk in the folder `./pipeline_cache`.
We do not expect uses to interact directly with this folder and it's contents should not be altered.

```{r}
#| echo: false
fs::dir_tree(file.path(dir, ".pipeline_cache"), all = TRUE)
```

### Changing things

Let's assume that after cleaning our data we wanted to sample rows rather than take as is.
Let's add a new function to our `R` directory:
	
```{.r filename="R/sample_clean.R"}
sample_clean <- function(clean) {
    clean[sample(nrow(clean), replace = TRUE), ]
}
```
	
```{r}
#| eval: false
cat(
    "sample_clean <- function(clean) clean[sample(nrow(clean), replace = TRUE), ]",
    file = file.path("R", "sample_clean.R")
)
```
	
```{r}
#| echo: false
cat(
    "sample_clean <- function(clean) clean[sample(nrow(clean), replace = TRUE), ]",
    file = file.path(dir, "R", "sample_clean.R")
)
```

Updating our pipeline accordingly, we have

```{r}
#| eval: false
out <- pipeline_run({
    raw     <- load_dat(CONFIG$in_csv)
    clean   <- wrangle_dat(raw, CONFIG$rows)
    samples <- sample_clean(clean)
    plot    <- plot_dat(samples, CONFIG$out_plot)
})
```

 

```{r}
#| echo: false
out <- pipeline_run(
    {
        raw     <- load_dat(CONFIG$in_csv)
        clean   <- wrangle_dat(raw, CONFIG$rows)
        samples <- sample_clean(clean)
        plot    <- plot_dat(samples, CONFIG$out_plot)
    },
    pipeline_dir = dir
)
```

Note that where possible, rather than start from scratch, the cached results were loaded from disk.
This caching takes account of changes in functions within the function directory but does not attempt to handle changes in the underlying R version or packages you use
(in essence, it assumes environment stability is managed elsewhere).


## Multiple scenarios

The package assists in running multiple scenarios in two ways.
Firstly, users can specify different configurations to run in the call, i.e.

```{r}
#| eval: false
pipeline_run(
    ...,
    config = c("default", "production"),
    scenario_default = "default"
)
```
This in turn tells the function to parse the configuration file, looking for an entry called 'default'.
This is used as the base scenario.
Other values (in this case 'production') in the configuration file are then layered on top of this (internally using `modifyList()`) to generate additional scenarios to run.

Secondly, we want to avoid unnecessary computations so, where possible, we try to share cached objects across scenarios.
To make things more concrete, consider our second demo:

```{r}
#| eval: false
dir2 <- pipeline_skeleton(dir = tempfile(), single = FALSE)
setwd(dir2)
```

```{r}
#| echo: false
f2 <- tempfile()
dir2 <- tryCatch(
    pipeline_skeleton(tempfile(), single = FALSE),
    error = function(cond) {
        unlink(2)
        stop("Unable to create pipeline skeleton.")
    }
)
```

This time the configuration file defines two scenarios, *default*, which subsets the data to 10 rows and *production*, which returns 32 rows.
Note that when working across different scenarios, the entries in the configuration file must be appropriately named lists.

::: {.callout-example data-legend="config.R"}
```{.r}
default <- list(
    in_csv   = INFILE("data/mtcars.csv"),
    out_plot = OUTFILE("plot.png"),
    rows     = 10
)

production <- list(
    rows = 32
)
```
:::

To run both scenarios:

```{r}
#| eval: false
out <- pipeline_run(
    {
        raw <- load_dat(CONFIG$in_csv)
        clean <- wrangle_dat(raw, CONFIG$rows)
        plot <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = c("default", "production"),
    scenario_default = "default" # included here but note that this is
                                 # also the default value for the argument.
)
```

```{r}
#| echo: false
out <- pipeline_run(
    {
        raw <- load_dat(CONFIG$in_csv)
        clean <- wrangle_dat(raw, CONFIG$rows)
        plot <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = c("default", "production"),
    scenario_default = "default",
    pipeline_dir = dir2
)
```

We see above that the *default* scenario ran from scratch where as the *production* scenario was able to skip the first computation, loading the cached result from disk.

```{r}
#| echo: false
#| fig.cap: "The resulting plots in the 'output' directory."
#| fig.alt: "Two side by side box plots of cyl (x-axis) versus mpg (y-axis). The left-hand one represents the 'default' configuration, the right-hand, 'production'. There is a downward trend showing in mpg as cyl increases." # nolint: line_length_linter.
layout(matrix(c(1, 2), nrow = 1))
ylim <- c(0, max(out$production$clean$mpg) + 2)
plot(
    factor(out$default$clean$cyl), out$default$clean$mpg,
    ylim = ylim, xlab = "cyl", ylab = "mpg", main = "default"
)
plot(
    factor(out$production$clean$cyl), out$production$clean$mpg,
    ylim = ylim, xlab = "cyl", ylab = "mpg", main = "production"
)
```